Aff/Eff & Reactive Programming

[CK-LinoPro]

Hello,

I finally got around to looking into your approach of handling side-effects using Eff<T> & Aff<T>.
I really like this approach but I wonder how one could integrate it into an interactive workflow.

My approach so far is to have an object-oriented structure at an application level, split into different areas.
E.g. for WPF with MVVM approach, I'd have a view layer, a view model layer and a service layer.
The service layer then provides object-oriented interfaces for functional implementations, using primarily Atom<T> and Ref<T> to manage state.
Services are managed via DI.
Information exchange uses primarily reactive structures (e.g. Rx) with a push-based approach.

Your approach using Eff<T> & Aff<T> basically makes the entire service layer & DI obsolete which is great.
However, to manage a responsive and interactive UI application (without implementing MVU) I don't see how to get rid of event processing, preferably with a reactive approach.

Can you think of a simple way to unite Eff<T> & Aff<T> with reactive programming on an application-level?

[louthy]

Are you aware of the LanguageExt.Rx extensions? They allow you to wrap up an IObservable and have it consumed with an Aff.

Another way would be to build a rudimentary actor-model using the Pipes streaming capability:

using LanguageExt;
using LanguageExt.Sys;
using LanguageExt.Pipes;
using LanguageExt.Sys.Live;
using System.Reactive.Linq;
using LanguageExt.Sys.Traits;
using LanguageExt.Effects.Traits;
using static LanguageExt.Prelude;
using static LanguageExt.Pipes.Proxy;

public record Actor<RT, A>(Func<A, Eff<RT, Unit>> Post, Eff<Unit> Shutdown)
    where RT : struct, HasCancel<RT>;

public static class Actor<RT, S, A>
    where RT : struct, HasCancel<RT>
{
    public static Eff<RT, Actor<RT, A>> Observe(IObservable<A> stream, S initial, Func<S, A, Aff<RT, S>> inbox)
    {
        var state = Atom(initial);
        var queue = Queue<RT, A>();
        var items = Producer.merge(queue, observe(stream));
        return from cancel in fork(items | Message(state, inbox))
               select new Actor<RT, A>(queue.EnqueueEff, cancel);
    }
    
    public static Eff<RT, Actor<RT, A>> Spawn(S initial, Func<S, A, Aff<RT, S>> inbox)
    {
        var state = Atom(initial);
        var queue = Queue<RT, A>();
        return from cancel in fork(queue | Message(state, inbox))
               select new Actor<RT, A>(queue.EnqueueEff, cancel);
    }

    static Consumer<RT, A, Unit> Message(Atom<S> state, Func<S, A, Aff<RT, S>> inbox) =>
        from m in awaiting<A>()
        from _ in state.SwapAff(s => inbox(s, m))
        select unit;
}

That has two construction functions: Spawn and Observe:

• Spawn when run will return a Actor<RT, A> which allows posting of messages to its inbox via the Post effect. Also comes with a Shutdown effect.
• Observe is the same, but it will also listen to an IObservable stream

The inbox manages a state Atom (technically it doesn't need to be an Atom as the Queue system will make sure only one message is processed at a time, but it simplifies the mutation)

You could imagine a simple application then that launches a number of actors that listen to the outside world and manage 'global' state accordingly.

public static class App<RT>
    where RT : struct, 
        HasCancel<RT>,
        HasConsole<RT>
{
    // Entry point
    public static Eff<RT, Unit> Main =>
        from svc in Actor<RT, MousePos, MousePos>.Observe(MouseStream, MousePos.Zero, WhereIsMyMouse)
        from __1 in Console<RT>.readKey
        from __2 in svc.Shutdown
        from __3 in Console<RT>.writeLine("Finished")
        select unit;

    // Simple service that handles the event (message) and returns a new state
    static Aff<RT, MousePos> WhereIsMyMouse(MousePos state, MousePos message) =>
        from _ in Console<RT>.writeLine($"last: {state}, this: {message}")
        select message;
    
    // Example observable stream - mocking a mouse position stream
    public static readonly IObservable<MousePos> MouseStream =
        Observable.Zip(
                Observable.Range(0, 10),
                Observable.Range(0, 10))
            .Select(p => new MousePos(p[0], p[1]));
}

You can then just run your app with the runtime of your choice:

App<Runtime>.Main.Run(Runtime.New());

Does that help?

[CK-LinoPro]

Thank you, I think I understand the idea.
I'll have to think about it and play a round with it for a while to see how it scales.

[louthy]

@CK-LinoPro Bit of an early heads up: for language-ext v5 there will be first-class support for observables and enumerables in Eff and Aff expressions. For example:

   // Effect that combines both an IObservable and an Eff in the same expression.
   // The IObservable is automatically lifted into the Eff, which causes the remainder of the
   // LINQ expression to be part of the observable stream.
   // The effect could be forked, which would allow the parent expression to continue whilst 
   // the observable keeps listening.
   var effect = from s in Observable.Interval(TimeSpan.FromSeconds(1)).Take(5);
                from x in SuccessEff<Runtime, int>(10)
                select s * x;

   // Running this as an Eff without forking will cause this thread to block until the observable
   // completes or errors.
   var result = effect.Run(Runtime.New());

   Console.WriteLine(result);

That will produce:

   [0, 10, 20, 30, 40]

Behind the scenes it will be using a new DSL based around ideas from category-theory. It will be a little bit like a toolkit for building composable types, like functors, monads, etc. I'll be retrofitting most of the complex monadic types to use this new DSL approach. That means these capabilities will be shared across more than just Aff and Eff.

It's still a long way off, even to the first beta, so don't hang too much on this. I just figured it'd be interesting to show the direction of travel.

[CK-LinoPro]

This looks awesome! I'm excited to see it in action one day. Thank you for your effort!