Finite yielding in side effecting source

[cheerio-pixel]

I don't know if is the right way of using pipes, but I have this pipe

    public static PipeT<Socket, Slice<byte>, M, Unit> receive(int bufferLength) =>
        from client in PipeT.awaiting<M, Socket, Slice<byte>>()
        from _ in PipeT.yieldRepeatIO<M, Socket, Slice<byte>>(
                    IO.liftAsync(async () =>
                        {
                            var buffer = new byte[bufferLength];
                            var length = await client.ReceiveAsync(buffer);
                            return new Slice<byte>(length, 0, buffer);
                        })
                )
        select unit;

So basically I want to stop yielding when the Slice has lenght zero (since that means that the client closed on us) but I don't see anything exposed publicly that allows me to fulfill this use case and I have checked the source code and see that I could use SourceT but to me looks like is not going to dispose of action of IO and is going to keep producing empty slices just that they aren't going to be pushed downstream. In this case what would be the best thing to do or is this just forcing pipes to do something is not supposed to do.

[louthy]

I'm unable to test this out, but something like this should work:

public static PipeT<Socket, Slice<byte>, M, Unit> receive(int bufferLength) =>
    receiveAny(bufferLength) | whileNotEmpty;

static PipeT<Socket, Slice<byte>, M, Unit> receiveAny(int bufferLength) =>
    from client in PipeT.awaiting<M, Socket, Slice<byte>>()
    from _      in PipeT.yieldRepeatIO<M, Socket, Slice<byte>>(read(client, bufferLength))
    select unit;

static readonly PipeT<Slice<byte>, Slice<byte>, M, Unit> whileNotEmpty =
    from slice in PipeT.awaiting<M, Slice<byte>, Slice<byte>>()
    from _     in slice.Length == 0
                      ? Pure(unit)
                      : PipeT.yield<M, Slice<byte>, Slice<byte>>(slice)
    select unit;

static IO<Slice<byte>> read(Socket client, int bufferLength) =>
    IO.liftAsync(async () =>
                 {
                     var buffer = new byte[bufferLength];
                     var length = await client.ReceiveAsync(buffer);
                     return new Slice<byte>(length, 0, buffer);
                 });

[louthy]

Also, there's some background info on the different streaming capabilities in the API documentation.

[cheerio-pixel]

To confirm, pipes, like the haskell implementation, is pull based, so if we stopped awaiting and sent a Pure instead (like in this example) it should terminate that pipe. Am I understanding correctly?

[cheerio-pixel]

I don't know if I'm doing things incorrectly but if do

    static IO<Slice<byte>> read(Socket client, int bufferLength) =>
        IO.liftAsync(async () =>
                     {
                         var buffer = new byte[bufferLength];
                         var length = await client.ReceiveAsync(buffer);
                         Console.WriteLine(length);
                         return new Slice<byte>(length, 0, buffer);
                     });

After closing the connection the consoles keeps writing 0 which I assume means that the pipe hasn't stopped running the IO. I kind of gave up for the moment and did this workaround to stop the whole pipeline

    static IO<Slice<byte>> read(Socket client, int bufferLength) =>
        IO.liftAsync(async () =>
                     {
                         var buffer = new byte[bufferLength];
                         var length = await client.ReceiveAsync(buffer);
                         if (length == 0)
                         {
                             throw new Exception();
                         }
                         return new Slice<byte>(length, 0, buffer);
                     });

Feels impure and doesn't show a stack trace but it seems to work

[louthy]

To confirm, pipes, like the haskell implementation, is pull based, so if we stopped awaiting and sent a Pure instead (like in this example) it should terminate that pipe. Am I understanding correctly?

Haskell Pipes supports both push and pull models, it really depends on the Producer and the Consumer to what the exact behaviour will be. Language-Ext Pipes used to be an exact clone of Haskell Pipes -- well, almost, the old version could only lift Eff<RT, A> into its transformer stack, but other than that the behaviour was a 1-to-1 match.

The new version of Language-Ext Pipes is fully generalised to use any M constrained to MonadIO but also the whole library has been completely written to be 'C# first'. I took out the bi-directional nature of pipes, because I see less value in that, and I tried to make it much easier to use in C# land.

That means Language-Ext Pipes and Haskell Pipes don't necessarily have exactly matching behaviours.

I wrote some relatively comprehensive release notes at the time about the changes.

The internals are still an ADT based DSL, but with more cases to support common C# patterns, namely the built-in lazy processing of IEnumerable and IAsyncEnumerable. One is pull (IEnumerable) and one is push (IAsyncEnumerable). It also bakes in concurrency support (which is a separate library in Haskell Pipes Concurrency). We can now directly lift a Source or SourceT into language-ext Pipes and are able to lift asynchronous behaviours.

All of that is a pragmatic choice to making Pipes easier to use in C#. Although they are still not quite as friendly to use as say Conduit, ConduitT, Source, or SourceT, which are more 'classic' forms of streaming.

So, back to your question:

so if we stopped awaiting and sent a Pure instead (like in this example) it should terminate that pipe. Am I understanding correctly?

No, I should have spotted that, it will terminate the piping of that value (stop it flowing downstream), but it won't stop subsequent values being pushed or pulled downstream from upstream sources.

The pipe gets 'compiled' into whatever monad-type you lift into it. So, for IO, you end up with an IO<A> monad instance when running the pipe (which you then subsequently Run to produce the effect). So, to terminate the pipe, you need to terminate the IO monad.

Let's look at the different approaches:

var prod = yieldAll<IO, int>(Range(1, 100));

var pipe = from x in awaiting<IO, int, int>()
           from _ in x == 10
                         ? Pure(unit)
                         : yield<IO, int, int>(x)
           select unit;

var cons = from x in awaiting<IO, int>()
           from _ in writeLine(x)
           select unit;

var effect = prod | pipe | cons;

var result = effect.Run().Run();

static IO<Unit> writeLine(object? value) =>
    IO.lift(() => Console.Write($"{value} "));

This will produce:

1 2 3 4 5 6 7 8 9 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100

Notice how 10 is missing from the list. So, we stopped it flowing downstream, but we didn't stop subsequent upstream values flowing down.

Next we use the PipeT.error command that works with lifted monads that are Fallible<M>. There's also PipeT.fail that works for Fallible<E, M> traits. You don't have to use these functions, you can just lift whatever monad you're working with into the pipe in its 'alternative state' and it will work the same...

var prod = yieldAll<IO, int>(Range(1, 100));

var pipe = from x in awaiting<IO, int, int>()
           from _ in x == 10
                         ? error<int, int, IO, Unit>(Errors.EndOfStream)
                         : yield<IO, int, int>(x)
           select unit;

var cons = from x in awaiting<IO, int>()
           from _ in writeLine(x)
           select unit;

var effect = prod | pipe | cons;

This prints out:

1 2 3 4 5 6 7 8 9 Unhandled exception. end of stream

So, this does what you want in that it terminates the subsequent processing of the upstream values.

IO uses exceptions internally, which is why we see Unhandled Exception. Let's catch that error so it doesn't yield an exception:

var prod = yieldAll<IO, int>(Range(1, 100));

var pipe = from x in awaiting<IO, int, int>()
           from _ in x == 10
                         ? error<int, int, IO, Unit>(Errors.EndOfStream)
                         : yield<IO, int, int>(x)
           select unit;

var cons = from x in awaiting<IO, int>()
           from _ in writeLine(x)
           select unit;

var effect = prod | pipe | cons;

var io = effect.Run().As() | unit;

var result = io.Run();

That prints out:

1 2 3 4 5 6 7 8 9 

Obviously, that swallows all errors, so instead of effect.Run().As() | unit you can use @catch to catch specific expected errors that are used to terminate the stream:

var prod = yieldAll<IO, int>(Range(1, 100));

var pipe = from x in awaiting<IO, int, int>()
           from _ in x == 10
                         ? error<int, int, IO, Unit>(Errors.EndOfStream)
                         : yield<IO, int, int>(x)
           select unit;

var cons = from x in awaiting<IO, int>()
           from _ in writeLine(x)
           select unit;

var effect = prod | pipe | cons;

var io = effect.Run()
            | @catch(Errors.EndOfStream, IO.pure(unit));

var result = io.Run();

That catches EndOfStream errors only and lets all others propagate. I have a wiki article on error handling which is worth a look also.

Then you can wrap up the end-of-stream error:

public static PipeT<IN, OUT, IO, Unit> endOfStream<IN, OUT>() =>
    error<IN, OUT, IO, Unit>(Errors.EndOfStream);

And create an extension method to catch the error when its run:

public static class EffectExtensions
{
    public static K<M, Unit> RunToEnd<M>(this EffectT<M, Unit> effect)
        where M : MonadIO<M>, Fallible<M> =>
        effect.Run()
      | @catch(Errors.EndOfStream, pure<M, Unit>(unit));
}

Which makes the code a little more concise:

var pipe = from x in awaiting<IO, int, int>()
           from _ in x == 10
                         ? endOfStream<int, int>()
                         : yield<IO, int, int>(x)
           select unit;

And:

var result = effect.RunToEnd().Run();

Finally, there is empty:

var prod = yieldAll<IO, int>(Range(1, 100));

var pipe = from x in awaiting<IO, int, int>()
           from _ in x == 10
                         ? empty<int, int, IO, Unit>()
                         : yield<IO, int, int>(x)
           select unit;

var cons = from x in awaiting<IO, int>()
           from _ in writeLine(x)
           select unit;

var effect = prod | pipe | cons;

This has a similar effect to Pure in that it's the equivalent of yieldAll with an empty sequence.

In your comments you mentioned throwing an Exception. That's fine for exceptional events and internally IO uses C# exceptions to make the flowing of errors a bit more efficient. But, I tend to find that stepping out of the lifted IO function and using IO.fail, Fail(error), or the fact that Error implicitly converts to IO<A> 'feels' better for non-exceptional errors (even if the underlying result is the same):

static IO<Slice<byte>> read(Socket client, int bufferLength) =>
    from s in IO.liftAsync(async () =>
                           {
                               var buffer = new byte[bufferLength];
                               var length = await client.ReceiveAsync(buffer);
                               return new Slice<byte>(length, 0, buffer);
                           })
    from r in s.Length == 0
                  ? Errors.EndOfStream
                  : IO.pure(s)
    select r;

[louthy]

@cheerio-pixel I notice you marked it as Answered as I was doing some edits. I added some additional comments about wrapping up end-of-stream error and the catching just the end-of-stream error so it doesn't propagate. I hope that helps.