PipeExample.cs

interface IPipe
{
    Type StateType { get; }
    Type[] InputMessageTypes { get; }
    IPipeOutput Transform(object message, object state);
}

public interface IPipeOutput
{
    object[] Messages { get; }
    object State { get; }
}

interface IMemory
{
    Type StateType { get; }
    Type[] InputMessageTypes { get; }
    IMemoryAccess Access(object message);
}

interface IMemoryAccess
{
    Func<Task<object>> Get { get; }
    Func<object, Task> Save { get; }
}

interface IBus
{
    Task Publish(object message);
}

async Task Main()
{
    // Fake having memory, pipe, message and bus. This will be tied together by the plumber.
    var memory = (IMemory)(new object());
    var pipe = (IPipe)(new object());
    var inputMessage = new object();
    var bus = (IBus)(new object());

    // Extract state getter and setter from the memory, based on the message being handled.
    var memoryAccess = memory.Access(inputMessage);

    var state = await memoryAccess.Get();
    var output = pipe.Transform(inputMessage, state);
    await memoryAccess.Save(output.State); // Could throw concurrency conflict exception. Let the bus retry delivery of message.
    foreach (var outputMessage in output.Messages)
    {
        await bus.Publish(outputMessage);
    }
}