yieldAll with source breaks consumer IO

[djm132]

Hi,

Found this one with strange behavior depending on how producer was created. Case B hangs on IO, works without IO. Both sync and async producers hangs on Case B, and works on A.

Case A

When producer defined as

yieldAll<Eff<RT>, int>(RndFeedSync()); // Using IEnumerable directly

consumer IO works.

Case B

When producer defined as

yieldAll<Eff<RT>, int>(RndFeedSync().AsSource()); // Using IEnumerable wrapped by Source

Expected behavior

Consumer should work with any producer and should not rely on producer creation details (probably far away in code).

Full test code

using LanguageExt;
using LanguageExt.Pipes;
using System.Runtime.CompilerServices;
using static LanguageExt.Pipes.ConsumerT;
using static LanguageExt.Pipes.ProducerT;
using static LanguageExt.Prelude;

namespace Test;

/// <summary>
/// Pipes issue.
/// </summary>
public class PipesIssue<RT> where RT : notnull
{
    // Async source
    static async IAsyncEnumerable<int> RndFeedAsync([EnumeratorCancellation] CancellationToken ct = default)
    {
        while (true)
        {
            yield return Random.Shared.Next(1, 100);
            await Task.Delay(100, ct);
        }
    }

    // Sync source
    static IEnumerable<int> RndFeedSync()
    {
        while (true)
        {
            yield return Random.Shared.Next(1, 100);
            Task.Delay(100).GetAwaiter().GetResult();
        }
    }

    // Producer
    private static ProducerT<int, Eff<RT>, Unit> prod =>
        //yieldAll<Eff<RT>, int>(RndFeedSync());                    // This one works
        yieldAll<Eff<RT>, int>(RndFeedAsync().AsSource());          // This one hangs if consumer has IO

    // Consumer
    private static ConsumerT<int, Eff<RT>, Unit> cons =>
        from v in awaiting<Eff<RT>, int>().Map(x =>
        {
            Console.WriteLine($"received {x}");
            return x;
        })
        from _ in IO.lift(() => Console.WriteLine($"IO got {v}"))   // Consumer hangs here in Case B
        select unit;

    // Test
    public static Eff<RT, Unit> pipeTest =>
        from _1 in IO.lift(() => Console.WriteLine("Starting pipe test"))
        from _2 in (prod | cons).Run().As()
        select unit;
}