Producing effects from IAsyncEnumerable

[ohurvitz]

I’m trying to use CliWrap to run a command and send the output to a server.
I’m trying to have code that looks like:

from cmd in liftEff(() => Cli.Wrap("ls").WithArguments("-la"))
from evnt in [something here](cmd.ListenAsync())
from _ in [create an effect to send output to the server]

But I’m not sure how to do this - or even if this makes sense.

[micmarsh]

I have basically no hands-on expertise with this, but this sounds like a use case for the (experimental) StreamT type (which I believe is being revised in the streaming branch?), or the Pipes system #1444.

[louthy]

I have basically no hands-on expertise with this, but this sounds like a use case for the (experimental) StreamT type (which I believe is being revised in the streaming branch?), or the Pipes system

This is true, but it's not quite ready yet. The Eff type is not a stream-type, it only yields singular values. The only currently official way is with pipes; but pipes is in a state of flux in v5. I think you're using v5, so avoid it until I do the next major release which is all about working with streams.

In the meantime, you can however use IteratorAsync, which wraps up IAsyncEnumerable. IteratorAsync turns the IAsyncEnumerable interface into something that can be used with functional behaviours.

To make it work with Eff you'll need some static functions that can lift it up:

public static Eff<bool> isEmpty<E>(IteratorAsync<E> iter) =>
    liftEff(async _ => await iter.IsEmpty);

public static Eff<E> head<E>(IteratorAsync<E> iter) =>
    liftEff(async _ => await iter.Head);

public static Eff<IteratorAsync<E>> tail<E>(IteratorAsync<E> iter) =>
    liftEff(async _ => await iter.Split().Tail);

Then you can write a function that recursively iterates the sequence:

public static Eff<Unit> listen(IteratorAsync<Event> iter) =>
    isEmpty(iter)
       .Bind(f => f ? unitEff
                    : head(iter)
                        .Bind(output)
                        .Action(tail(iter).Bind(listen))); 

This is a common pattern in functional-programming: we check if the sequence is empty, if it is then we return; if not, then we read the head-element, process it, and then recursively call ourselves with the tail. It's not so common in C#-land because C# doesn't handle recursion well. We get stack-overflows if we go too deep.

However, Eff supports recursion (because it's based on IO, which has an internal state-machine that handles recursion properly -- so anything that leverages the IO monad, gets recursion for free).

You need to provide a function to process the value:

public static Eff<Unit> output(Event evnt) =>
    throw new NotImplementedException();

Then we can update your example to this:

public static Eff<Unit> go =>
    from cmd   in liftEff(() => Cli.Wrap("ls").WithArguments("-la"))
    from _     in listen(cmd.ListenAsync().GetIteratorAsync())
    select unit;

That should work pretty well.

Once the new streaming features are released then you'll be able to use SourceT:

public static Eff<Unit> go =>
    from cmd   in liftEff(() => Cli.Wrap("ls").WithArguments("-la"))
    from _     in listen(cmd).Iter()
    select unit;

public static SourceT<Eff, Unit> listen(Cmd cmd) =>
    from x in SourceT.lift<Eff, Event>(cmd.ListenAsync())
    from _ in output(x)
    select unit;

[ohurvitz]

Thank you so much. I have used some functional languages before (Clojure, Gleam) but am now trying to really get into it. And getting a C# team to use F# is a non-starter, and C# <-> F# interop is not great. I really want to fully understand effect systems. I have many years experience building complex distributed systems but always in imperative style - and I’m well aware of the limitations of that style. If you don’t mind me asking: is there some approximate timeline for v5 being more stable? I don’t need it to be released but want to try and get into more idiomatic use of language-ext.

On April 13, 2025 at 6:23 PM, Paul Louth ***@***.***) wrote: > I have basically no hands-on expertise with this, but this sounds like > a use case for the (experimental) StreamT type (which I believe is being > revised in the streaming branch?), or the Pipes system

This is true, but it's not quite ready yet. The Eff type is not a

…

stream-type, it only yields singular values. The only currently official way is with pipes; but pipes is in a state of flux in v5. I think you're using v5, so avoid it until I do the next major release which is all about working with streams. In the meantime, you can however use IteratorAsync, which wraps up IAsyncEnumerable. IteratorAsync turns the IAsyncEnumerable interface into something that can be used with functional behaviours. To make it work with Eff you'll need some static functions that can lift it up: public static Eff<bool> isEmpty<E>(IteratorAsync<E> iter) => liftEff(_ => iter.IsEmpty.AsTask()); public static Eff<E> head<E>(IteratorAsync<E> iter) => liftEff(_ => iter.Head.AsTask()); public static Eff<IteratorAsync<E>> tail<E>(IteratorAsync<E> iter) => liftEff(_ => iter.Split().Tail.AsTask()); Then you can write a function that recursively iterates the sequence: public static Eff<Unit> listen(IteratorAsync<Event> iter) => isEmpty(iter) .Bind(f => f ? unitEff : head(iter) .Bind(output) .Action(tail(iter).Bind(listen))); This is a common pattern in functional-programming: we check if the sequence is empty, if it is then we return; if not, then we read the head-element, process it, and then recursively call ourselves with the tail. It's not so common in C#-land because C# doesn't handle recursion well. We get stack-overflows if we go too deep. However, Eff supports recursion (because it's based on IO, which has an internal state-machine that handles recursion properly -- so anything that leverages the IO monad, gets recursion for free). You need to provide a function to process the value: public static Eff<bool> output(Event evnt) => throw new NotImplementedException(); Then we can update your example to this: public static Eff<Unit> go => from cmd in liftEff(() => Cli.Wrap("ls").WithArguments("-la")) from _ in listen(cmd.ListenAsync().GetIteratorAsync()) select unit; That should work pretty well. Once the new streaming features are released then you'll be able to use SourceT: public static Eff<Unit> go => from cmd in liftEff(() => Cli.Wrap("ls").WithArguments("-la")) from _ in listen(cmd).Iter() select unit; public static SourceT<Eff, Unit> listen(Cmd cmd) => from x in SourceT.lift<Eff, Event>(cmd.ListenAsync()) from _ in output(x) select unit; — Reply to this email directly, view it on GitHub <#1462 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABZFD47AHL6PZZZV6Q3FK4T2ZLPVNAVCNFSM6AAAAAB25E4S2CVHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTEOBSGI2DKOI>. You are receiving this because you authored the thread.Message ID: ***@***.***>

[louthy]

If you don’t mind me asking: is there some approximate timeline for v5
being more stable?

The streaming functionality 'kinda' is the last major job of the big v4 -> v5 10 year refactor. Unfortunately, one issue opened a Pandora's box of changes, which I figured I should spend the time upfront on. So, that's holding me up at the moment; but the positive is that the new streaming capabilities are gonna be extremely powerful.

After that I need to tidy up the Parsec project, it's been slightly neglected and still has some nullability warnings that needs consideration, and then it's finalising some decent unit-test coverage for the new functionality.

So, really, outside of the streaming functionality (which will land very soon), v5 is API-stable. And, right now, there are very few outstanding bugs -- although I can't discount the fact that the new unit-tests may unearth new issues.

I don't like putting fixed timelines on this project, as it's something I work on in my spare time, but I'd expect the streaming functionality to land before the end of April. The rest of it might take a bit longer. Knowing my pace on this, I'd say v5 could be RTM by July. But take these estimates with a pinch of salt.

If you're starting working with language-ext for the first time, start using v5 now. The outstanding issues are minor/have workarounds/are in v4 too/or are in new functionality that didn't even exist in v4.