MessagePipe

MessagePipe

MessagePipe is a high-performance in-memory/distributed messaging pipeline for .NET and Unity. It supports all cases of Pub/Sub usage, mediator pattern for CQRS, EventAggregator of Prism(V-VM decoupling), IPC(Interprocess Communication)-RPC, etc.

• Dependency-injection first
• Filter pipeline
• better event
• sync/async
• keyed/keyless
• buffered/bufferless
• singleton/scoped
• broadcast/response(+many)
• in-memory/interprocess/distributed

MessagePipe is faster than standard C# event and 78 times faster than Prism's EventAggregator.

Of course, memory allocation per publish operation is less(zero).

Also providing roslyn-analyzer to prevent subscription leak.

Getting Started

For .NET, use NuGet. For Unity, please read Unity section.

PM> Install-Package MessagePipe

MessagePipe is built on top of a Microsoft.Extensions.DependencyInjection(for Unity, VContainer or Zenject or Builtin Tiny DI) so set up via ConfigureServices in .NET Generic Host. Generic Host is widely used in .NET such as ASP.NET Core, MagicOnion, ConsoleAppFramework, MAUI, WPF(with external support), etc so easy to setup.

using MessagePipe;
using Microsoft.Extensions.DependencyInjection;

Host.CreateDefaultBuilder()
    .ConfigureServices((ctx, services) =>
    {
        services.AddMessagePipe(); // AddMessagePipe(options => { }) for configure options
    })

Get the IPublisher<T> for publisher, Get the ISubscribe<T> for subscriber, like a Logger<T>. T can be any type, primitive(int, string, etc...), struct, class, enum, etc.

…

It is similar to event, but decoupled by type as key. The return value of Subscribe is IDisposable, which makes it easier to unsubscribe than event. You can release many subscriptions at once by DisposableBag(CompositeDisposable). See the Managing Subscription and Diagnostics section for more details.

The publisher/subscriber(internally we called MessageBroker) is managed by DI, it is possible to have different broker for each scope. Also, all subscriptions are unsubscribed when the scope is disposed, which prevents subscription leaks.

Default is singleton, you can configure MessagePipeOptions.InstanceLifetime to Singleton or Scoped.

IPublisher<T>/ISubscriber<T> is keyless(type only) however MessagePipe has similar interface IPublisher<TKey, TMessage>/ISubscriber<TKey, TMessage> that is keyed(topic) interface.

For example, our real usecase, There is an application that connects Unity and MagicOnion (a real-time communication framework like SignalR) and delivers it via a browser by Blazor. At that time, we needed something to connect Blazor's page (Browser lifecycle) and MagicOnion's Hub (Connection lifecycle) to transmit data. We also need to distribute the connections by their IDs.

Browser <-> Blazor <- [MessagePipe] -> MagicOnion <-> Unity

We solved this with the following code.

…

The main difference of Reactive Extensions' Subject is has no OnCompleted. OnCompleted may or may not be used, making it very difficult to determine the intent to the observer(subscriber). Also, we usually subscribe to multiple events from the same (different event type)publisher, and it is difficult to handle duplicate OnCompleted in that case. For this reason, MessagePipe only provides a simple Publish(OnNext). If you want to convey completion, please receive a separate event and perform dedicated processing there.

In other words, this is the equivalent of Relay in RxSwift.

In addition to standard Pub/Sub, MessagePipe supports async handlers, mediator patterns with handlers that accept return values, and filters for pre-and-post execution customization.

This image is a visualization of the connection between all those interfaces.

You may be confused by the number of interfaces, but many functions can be written with a similar, unified API.

Publish/Subscribe

Publish/Subscribe interface has keyed(topic) and keyless, sync and async interface.

…

All are available in the form of IPublisher/Subscribe<T> in the DI. async handler can await all subscribers completed by await PublishAsync. Asynchronous methods can work sequentially or in parallel, depending on AsyncPublishStrategy (defaults is Parallel, can be changed by MessagePipeOptions or by specifying at publish time). If you don't need to wait, you can call void Publish to act as fire-and-forget.

The before and after of execution can be changed by passing a custom filter. See the Filter section for details.

If an error occurs, it will be propagated to the caller and subsequent subscribers will be stopped. This behavior can be changed by writing a filter to ignore errors.

ISingleton***, IScoped***

I(Async)Publisher(Subscriber)'s lifetime belongs to MessagePipeOptions.InstanceLifetime. However if declare with ISingletonPublisher<TMessage>/ISingletonSubscriber<TKey, TMessage>, ISingletonAsyncPublisher<TMessage>/ISingletonAsyncSubscriber<TKey, TMessage> then used singleton lifetime. Also IScopedPublisher<TMessage>/IScopedSubscriber<TKey, TMessage>, IScopedAsyncPublisher<TMessage>/IScopedAsyncSubscriber<TKey, TMessage> uses scoped lifetime.

Buffered

IBufferedPublisher<TMessage>/IBufferedSubscriber<TMessage> pair is similar as BehaviorSubject or Reactive Extensions(More equal is RxSwift's BehaviorRelay). It returns latest value on Subscribe.

var p = provider.GetRequiredService<IBufferedPublisher<int>>();
var s = provider.GetRequiredService<IBufferedSubscriber<int>>();

p.Publish(999);

var d1 = s.Subscribe(x => Console.WriteLine(x)); // 999
p.Publish(1000); // 1000

var d2 = s.Subscribe(x => Console.WriteLine(x)); // 1000
p.Publish(9999); // 9999, 9999

DisposableBag.Create(d1, d2).Dispose();

If TMessage is class and does not have latest value(null), does not send value on Subscribe.

Keyed buffered publisher/subscriber does not exist because difficult to avoid memory leak of (unused)key and keep latest value.

EventFactory

Using EventFactory, you can create generic IPublisher/ISubscriber, IAsyncPublisher/IAsyncSubscriber, IBufferedPublisher/IBufferedSubscriber, IBufferedAsyncPublisher/IBufferedAsyncSubscriber like C# events, with a Subscriber tied to each instance, not grouped by type.

MessagePipe has better properties than a normal C# event

• Using Subscribe/Dispose instead of +=, -= , easy to management subscription
• Both sync and async support
• Both bufferless and buffered support
• Enable unsubscribe all subscription from publisher.dispose
• Attaches invocation pipeline behaviour by Filter
• To monitor subscription leak by MessagePipeDiagnosticsInfo
• TO prevent subscription leak by MessagePipe.Analyzer

…

If you want to create event outside of DI, see Global Provider section.

IDisposablePublisher<int> tickPublisher;
public ISubscriber<int> OnTick { get; }

ctor()
{
    (tickPublisher, OnTick) = GlobalMessagePipe.CreateEvent<int>();
}

Request/Response/All

Similar as MediatR, implement support of mediator pattern.

public interface IRequestHandler<in TRequest, out TResponse>
{
    TResponse Invoke(TRequest request);
}

public interface IAsyncRequestHandler<in TRequest, TResponse>
{
    ValueTask<TResponse> InvokeAsync(TRequest request, CancellationToken cancellationToken = default);
}

For example, declare handler for Ping type.

public readonly struct Ping { }
public readonly struct Pong { }

public class PingPongHandler : IRequestHandler<Ping, Pong>
{
    public Pong Invoke(Ping request)
    {
        Console.WriteLine("Ping called.");
        return new Pong();
    }
}

You can get handler like this.

class FooController
{
    IRequestHandler<Ping, Pong> requestHandler;

    // automatically instantiate PingPongHandler.
    public FooController(IRequestHandler<Ping, Pong> requestHandler)
    {
        this.requestHandler = requestHandler;
    }

    public void Run()
    {
        var pong = this.requestHandler.Invoke(new Ping());
        Console.WriteLine("PONG");
    }
}

For more complex implementation patterns, this Microsoft documentation is applicable.

Declare many request handlers, you can use IRequestAllHandler, IAsyncRequestAllHandler instead of single handler.

…

…

Subscribe Extensions

ISubscriber(IAsyncSubscriber) interface requires IMessageHandler<T> to handle message.

public interface ISubscriber<TMessage>
{
    IDisposable Subscribe(IMessageHandler<TMessage> handler, params MessageHandlerFilter<TMessage>[] filters);
}

However, the extension method allows you to write Action<T> directly.

…

Also, the Func<TMessage, bool> overload can filter messages by predicate (internally implemented with PredicateFilter, where Order is int.MinValue and is always checked first).

AsObservable can convert message pipeline to IObservable<T>, it can handle by Reactive Extensions(in Unity, you can use UniRx). AsObservable exists in sync subscriber(keyless, keyed, buffered).

AsAsyncEnumerable can convert message pipeline to IAsyncEnumerable<T>, it can handle by async LINQ and async foreach. AsAsyncEnumerable exists in async subscriber(keyless, keyed, buffered).

FirstAsync gets the first value of message. It is similar as AsObservable().FirstAsync(), AsObservable().Where().FirstAsync(). If uses CancellationTokenSource(TimeSpan) then similar as AsObservable().Timeout().FirstAsync(). Argument of CancellationToken is required to avoid task leak.

// for Unity, use cts.CancelAfterSlim(TIimeSpan) instead.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(1));
var value = await subscriber.FirstAsync(cts.Token);

FirstAsync exists in both sync and async subscriber(keyless, keyed, buffered).

Filter

Filter system can hook before and after method invocation. It is implemented with the Middleware pattern, which allows you to write synchronous and asynchronous code with similar syntax. MessagePipe provides different filter types - sync (MessageHandlerFilter<T>), async (AsyncMessageHandlerFilter<T>), request (RequestHandlerFilter<TReq, TRes>) and async request (AsyncRequestHandlerFilter<TReq, TRes>). To implement other concerete filters the above filter types can be extended.

Filters can be specified in three places - global(by MessagePipeOptions.AddGlobalFilter), per handler type, and per subscription. These filters are sorted according to the Order specified in each of them, and are generated when subscribing.

Since the filter is generated on a per subscription basis, the filter can have a state.

…

use the filter by attribute, you can use these attributes: [MessageHandlerFilter(type, order)], [AsyncMessageHandlerFilter(type, order)], [RequestHandlerFilter(type, order)], [AsyncRequestHandlerFilter(type, order)].

These are idea showcase of filter.

public class PredicateFilter<T> : MessageHandlerFilter<T>
{
    private readonly Func<T, bool> predicate;

    public PredicateFilter(Func<T, bool> predicate)
    {
        this.predicate = predicate;
    }

    public override void Handle(T message, A