MessagePipe Documentation

Filters implement the Middleware pattern, allowing you to hook into the invocation pipeline before and after a message is handled.

Filter Types

• MessageHandlerFilter<T> (Sync)
• AsyncMessageHandlerFilter<T> (Async)
• RequestHandlerFilter<TReq, TRes> (Request)
• AsyncRequestHandlerFilter<TReq, TRes> (Async Request)

Where to apply Filters

1. Global: Via MessagePipeOptions.AddGlobalMessageHandlerFilter during setup.
2. Per Handler Type: Using the [MessageHandlerFilter(type, order)] attribute on a handler class.
3. Per Subscription: Passing filter instances directly into the Subscribe method.

Filters are sorted by their Order property.

// 1. Define a custom filter
public class ChangedValueFilter<T> : MessageHandlerFilter<T>
{
    T lastValue;
    public override void Handle(T message, Action<T> next)
    {
        if (EqualityComparer<T>.Default.Equals(message, lastValue)) return;
        lastValue = message;
        next(message);
    }
}

// 2. Apply per subscription
subscriber.Subscribe(x => Console.WriteLine(x), new ChangedValueFilter<int>(){ Order = 100 });

// 3. Apply per handler type using attributes
[MessageHandlerFilter(typeof(ChangedValueFilter<>), 100)]
public class WriteLineHandler<T> : IMessageHandler<T>
{
    public void Handle(T message) => Console.WriteLine(message);
}

// 4. Apply globally during DI configuration
services.AddMessagePipe(options =>
{
    options.AddGlobalMessageHandlerFilter(typeof(ChangedValueFilter<>), 100);
});

You can perform Remote Procedure Calls (RPC) across processes using IRemoteRequestHandler<TRequest, TResponse> on the server side and IRemoteRequestHandler<TRequest, TResponse> on the client side. This requires TcpInterprocess or NamedPipeInterprocess to be enabled.

Server Configuration

Set HostAsServer = true in the interprocess options.

Implementation Example

Server Handler:

public class MyAsyncHandler : IAsyncRequestHandler<int, string>
{
    public async ValueTask<string> InvokeAsync(int request, CancellationToken cancellationToken = default)
    {
        await Task.Delay(1);
        return request == -1 ? throw new Exception("NO -1") : $"ECHO:{request}";
    }
}

Client Call:

async void A(IRemoteRequestHandler<int, string> remoteHandler)
{
    var v = await remoteHandler.InvokeAsync(9999);
    Console.WriteLine(v); // ECHO:9999
}

Host.CreateDefaultBuilder()
    .ConfigureServices((ctx, services) => {
        services.AddMessagePipe()
            .AddTcpInterprocess("127.0.0.1", 3215, x => {
                x.HostAsServer = true;
            });
    });

The EventFactory allows you to create generic publishers and subscribers (IPublisher/ISubscriber, IAsyncPublisher/IAsyncSubscriber, etc.) that behave like C# events but are tied to specific instances rather than grouped by type.

Key advantages over standard C# events:

• Use Subscribe/Dispose for easier subscription management.
• Supports both synchronous and asynchronous patterns.
• Supports both bufferless and buffered modes.
• Calling Dispose() on a publisher unsubscribes all associated subscribers.
• Supports an invocation pipeline via Filters.
• Provides diagnostics via MessagePipeDiagnosticsInfo and leak prevention via MessagePipe.Analyzer.

If you are working outside of Dependency Injection, you can use GlobalMessagePipe.CreateEvent<T>().

public class BetterEvent : IDisposable
{
    IDisposablePublisher<int> tickPublisher;
    public ISubscriber<int> OnTick { get; }

    public BetterEvent(EventFactory eventFactory)
    {
        // CreateEvent returns a tuple containing the publisher and the subscriber
        (tickPublisher, OnTick) = eventFactory.CreateEvent<int>();
    }

    int count;
    void Tick()
    {
        tickPublisher.Publish(count++);
    }

    public void Dispose()
    {
        // Unsubscribes all from the publisher
        tickPublisher.Dispose();
    }
}

If you are using BuiltinContainerBuilder, it does not support scopes (it is always InstanceScope.Singleton), IRequestAllHandler/IAsyncRequestAllHandler, or many other DI functionalities. In these cases, it is recommended to use GlobalMessagePipe to access publishers and subscribers.

To use this pattern:

1. Build your service provider.
2. Set it using GlobalMessagePipe.SetProvider(provider).
3. Access publishers and subscribers via GlobalMessagePipe.GetPublisher<T>() and GlobalMessagePipe.GetSubscriber<T>().

// create provider and set to Global(to enable diagnostics window and global fucntion)
var provider = builder.BuildServiceProvider();
GlobalMessagePipe.SetProvider(provider);

// --- to use MessagePipe, you can use from GlobalMessagePipe.
var p = GlobalMessagePipe.GetPublisher<int>();
var s = GlobalMessagePipe.GetSubscriber<int>();

var d = s.Subscribe(x => Debug.Log(x));

p.Publish(10);
p.Publish(20);
p.Publish(30);

d.Dispose();