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.

usingMessagePipe;usingMicrosoft.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.

usingMessagePipe;publicstructMyEvent{}publicclassSceneA{readonlyIPublisher<MyEvent>publisher;publicSceneA(IPublisher<MyEvent>publisher){this.publisher=publisher;}voidSend(){this.publisher.Publish(newMyEvent());}}publicclassSceneB{readonlyISubscriber<MyEvent>subscriber;readonlyIDisposabledisposable;publicSceneB(ISubscriber<MyEvent>subscriber){varbag=DisposableBag.CreateBuilder();// composite disposable for manage subscriptionsubscriber.Subscribe(x =>Console.WriteLine("here")).AddTo(bag);disposable=bag.Build();}voidClose(){disposable.Dispose();// unsubscribe event, all subscription **must** Dispose when completed}}

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.

// MagicOnion(similar as SignalR, realtime event framework for .NET and Unity)publicclassUnityConnectionHub:StreamingHubBase<IUnityConnectionHub,IUnityConnectionHubReceiver>,IUnityConnectionHub{readonlyIPublisher<Guid,UnitEventData>eventPublisher;readonlyIPublisher<Guid,ConnectionClose>closePublisher;Guidid;publicUnityConnectionHub(IPublisher<Guid,UnitEventData>eventPublisher,IPublisher<Guid,ConnectionClose>closePublisher){this.eventPublisher=eventPublisher;this.closePublisher=closePublisher;}overrideasyncValueTaskOnConnected(){this.id=Guid.Parse(Context.Headers["id"]);}overrideasyncValueTaskOnDisconnected(){this.closePublisher.Publish(id,newConnectionClose());// publish to browser(Blazor)}// called from Client(Unity)publicTask<UnityEventData>SendEventAsync(UnityEventDatadata){this.eventPublisher.Publish(id,data);// publish to browser(Blazor)}}// BlazorpublicpartialclassBlazorPage:ComponentBase,IDisposable{[Parameter]publicGuidID{get;set;}[Inject]ISubscriber<Guid,UnitEventData>UnityEventSubscriber{get;set;}[Inject]ISubscriber<Guid,ConnectionClose>ConnectionCloseSubscriber{get;set;}IDisposablesubscription;protectedoverridevoidOnInitialized(){// receive event from MagicOnion(that is from Unity)vard1=UnityEventSubscriber.Subscribe(ID, x =>{// do anything...});vard2=ConnectionCloseSubscriber.Subscribe(ID, _ =>{// show disconnected thing to view...subscription?.Dispose();// and unsubscribe events.});subscription=DisposableBag.Create(d1,d2);// combine disposable.}publicvoidDispose(){// unsubscribe event when browser is closed.subscription?.Dispose();}}

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.

// keyless-syncpublicinterfaceIPublisher<TMessage>{voidPublish(TMessagemessage);}publicinterfaceISubscriber<TMessage>{IDisposableSubscribe(IMessageHandler<TMessage>handler,paramsMessageHandlerFilter<TMessage>[]filters);}// keyless-asyncpublicinterfaceIAsyncPublisher<TMessage>{// async interface's publish is fire-and-forgetvoidPublish(TMessagemessage,CancellationTokencancellationToken=default(CancellationToken));ValueTaskPublishAsync(TMessagemessage,CancellationTokencancellationToken=default(CancellationToken));ValueTaskPublishAsync(TMessagemessage,AsyncPublishStrategypublishStrategy,CancellationTokencancellationToken=default(CancellationToken));}publicinterfaceIAsyncSubscriber<TMessage>{IDisposableSubscribe(IAsyncMessageHandler<TMessage>asyncHandler,paramsAsyncMessageHandlerFilter<TMessage>[]filters);}// keyed-syncpublicinterfaceIPublisher<TKey,TMessage>whereTKey:notnull{voidPublish(TKeykey,TMessagemessage);}publicinterfaceISubscriber<TKey,TMessage>whereTKey:notnull{IDisposableSubscribe(TKeykey,IMessageHandler<TMessage>handler,paramsMessageHandlerFilter<TMessage>[]filters);}// keyed-asyncpublicinterfaceIAsyncPublisher<TKey,TMessage>whereTKey:notnull{voidPublish(TKeykey,TMessagemessage,CancellationTokencancellationToken=default(CancellationToken));ValueTaskPublishAsync(TKeykey,TMessagemessage,CancellationTokencancellationToken=default(CancellationToken));ValueTaskPublishAsync(TKeykey,TMessagemessage,AsyncPublishStrategypublishStrategy,CancellationTokencancellationToken=default(CancellationToken));}publicinterfaceIAsyncSubscriber<TKey,TMessage>whereTKey:notnull{IDisposableSubscribe(TKeykey,IAsyncMessageHandler<TMessage>asyncHandler,paramsAsyncMessageHandlerFilter<TMessage>[]filters);}

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.

varp=provider.GetRequiredService<IBufferedPublisher<int>>();vars=provider.GetRequiredService<IBufferedSubscriber<int>>();p.Publish(999);vard1=s.Subscribe(x =>Console.WriteLine(x));// 999p.Publish(1000);// 1000vard2=s.Subscribe(x =>Console.WriteLine(x));// 1000p.Publish(9999);// 9999, 9999DisposableBag.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

publicclassBetterEvent:IDisposable{// using MessagePipe instead of C# event/Rx.Subject// store Publisher to private field(declare IDisposablePublisher/IDisposableAsyncPublisher)IDisposablePublisher<int>tickPublisher;// Subscriber is used from outside so public propertypublicISubscriber<int>OnTick{get;}publicBetterEvent(EventFactoryeventFactory){// CreateEvent can deconstruct by tuple and set together(tickPublisher,OnTick)=eventFactory.CreateEvent<int>();// also create async event(IAsyncSubscriber) by `CreateAsyncEvent`// eventFactory.CreateAsyncEvent}intcount;voidTick(){tickPublisher.Publish(count++);}publicvoidDispose(){// You can unsubscribe all from Publisher.tickPublisher.Dispose();}}

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

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

Request/Response/All

Similar as MediatR, implement support of mediator pattern.

publicinterfaceIRequestHandler<inTRequest,outTResponse>{TResponseInvoke(TRequestrequest);}publicinterfaceIAsyncRequestHandler<inTRequest,TResponse>{ValueTask<TResponse>InvokeAsync(TRequestrequest,CancellationTokencancellationToken=default);}

For example, declare handler for Ping type.

publicreadonlystructPing{}publicreadonlystructPong{}publicclassPingPongHandler:IRequestHandler<Ping,Pong>{publicPongInvoke(Pingrequest){Console.WriteLine("Ping called.");returnnewPong();}}

You can get handler like this.

classFooController{IRequestHandler<Ping,Pong>requestHandler;// automatically instantiate PingPongHandler.publicFooController(IRequestHandler<Ping,Pong>requestHandler){this.requestHandler=requestHandler;}publicvoidRun(){varpong=this.requestHandler.Invoke(newPing());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.

publicinterfaceIRequestAllHandler<inTRequest,outTResponse>{TResponse[]InvokeAll(TRequestrequest);IEnumerable<TResponse>InvokeAllLazy(TRequestrequest);}publicinterfaceIAsyncRequestAllHandler<inTRequest,TResponse>{ValueTask<TResponse[]>InvokeAllAsync(TRequestrequest,CancellationTokencancellationToken=default);ValueTask<TResponse[]>InvokeAllAsync(TRequestrequest,AsyncPublishStrategypublishStrategy,CancellationTokencancellationToken=default);IAsyncEnumerable<TResponse>InvokeAllLazyAsync(TRequestrequest,CancellationTokencancellationToken=default);}

publicclassPingPongHandler1:IRequestHandler<Ping,Pong>{publicPongInvoke(Pingrequest){Console.WriteLine("Ping1 called.");returnnewPong();}}publicclassPingPongHandler2:IRequestHandler<Ping,Pong>{publicPongInvoke(Pingrequest){Console.WriteLine("Ping1 called.");returnnewPong();}}classBarController{IRequestAllHandler<Ping,Pong>requestAllHandler;publicFooController(IRequestAllHandler<Ping,Pong>requestAllHandler){this.requestAllHandler=requestAllHandler;}publicvoidRun(){varpongs=this.requestAllHandler.InvokeAll(newPing());Console.WriteLine("PONG COUNT:"+pongs.Length);}}

Subscribe Extensions

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

publicinterfaceISubscriber<TMessage>{IDisposableSubscribe(IMessageHandler<TMessage>handler,paramsMessageHandlerFilter<TMessage>[]filters);}

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

publicstaticIDisposableSubscribe<TMessage>(thisISubscriber<TMessage>subscriber,Action<TMessage>handler,paramsMessageHandlerFilter<TMessage>[]filters)publicstaticIDisposableSubscribe<TMessage>(thisISubscriber<TMessage>subscriber,Action<TMessage>handler,Func<TMessage,bool>predicate,paramsMessageHandlerFilter<TMessage>[]filters)publicstaticIObservable<TMessage>AsObservable<TMessage>(thisISubscriber<TMessage>subscriber,paramsMessageHandlerFilter<TMessage>[]filters)publicstaticIAsyncEnumerable<TMessage>AsAsyncEnumerable<TMessage>(thisIAsyncSubscriber<TMessage>subscriber,paramsAsyncMessageHandlerFilter<TMessage>[]filters)publicstaticValueTask<TMessage>FirstAsync<TMessage>(thisISubscriber<TMessage>subscriber,CancellationTokencancellationToken,paramsMessageHandlerFilter<TMessage>[]filters)publicstaticValueTask<TMessage>FirstAsync<TMessage>(thisISubscriber<TMessage>subscriber,CancellationTokencancellationToken,Func<TMessage,bool>predicate,paramsMessageHandlerFilter<TMessage>[]filters)

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.usingvarcts=newCancellationTokenSource(TimeSpan.FromSeconds(1));varvalue=awaitsubscriber.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.

publicclassChangedValueFilter<T>:MessageHandlerFilter<T>{TlastValue;publicoverridevoidHandle(Tmessage,Action<T>next){if(EqualityComparer<T>.Default.Equals(message,lastValue)){return;}lastValue=message;next(message);}}// uses(per subscribe)subscribe.Subscribe(x =>Console.WriteLine(x),newChangedValueFilter<int>(){Order=100});// add per handler type(use generics filter, write open generics)[MessageHandlerFilter(typeof(ChangedValueFilter<>),100)]publicclassWriteLineHandler<T>:IMessageHandler<T>{publicvoidHandle(Tmessage)=>Console.WriteLine(message);}// add per globalHost.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{services.AddMessagePipe(options =>{options.AddGlobalMessageHandlerFilter(typeof(ChangedValueFilter<>),100);});});

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.

publicclassPredicateFilter<T>:MessageHandlerFilter<T>{privatereadonlyFunc<T,bool>predicate;publicPredicateFilter(Func<T,bool>predicate){this.predicate=predicate;}publicoverridevoidHandle(Tmessage,Action<T>next){if(predicate(message)){next(message);}}}

publicclassLockFilter<T>:MessageHandlerFilter<T>{readonlyobjectgate=newobject();publicoverridevoidHandle(Tmessage,Action<T>next){lock(gate){next(message);}}}

publicclassIgnoreErrorFilter<T>:MessageHandlerFilter<T>{readonlyILogger<IgnoreErrorFilter<T>>logger;publicIgnoreErrorFilter(ILogger<IgnoreErrorFilter<T>>logger){this.logger=logger;}publicoverridevoidHandle(Tmessage,Action<T>next){try{next(message);}catch(Exceptionex){logger.LogError(ex,"");// error logged, but do not propagate}}}

publicclassDispatcherFilter<T>:MessageHandlerFilter<T>{readonlyDispatcherdispatcher;publicDispatcherFilter(Dispatcherdispatcher){this.dispatcher=dispatcher;}publicoverridevoidHandle(Tmessage,Action<T>next){dispatcher.BeginInvoke(()=>{next(message);});}}

publicclassDelayRequestFilter:AsyncRequestHandlerFilter<int,int>{publicoverrideasyncValueTask<int>InvokeAsync(intrequest,CancellationTokencancellationToken,Func<int,CancellationToken,ValueTask<int>>next){awaitTask.Delay(TimeSpan.FromSeconds(request));varresponse=awaitnext(request,cancellationToken);returnresponse;}}

Managing Subscription and Diagnostics

Subscribe returns IDisposable; when call Dispose then unsubscribe. A better reason than event is that it is easy to Unsubscribe. To manage multiple IDisposable, you can use CompositeDisposable in Rx(UniRx) or DisposableBag included in MessagePipe.

IDisposabledisposable;voidOnInitialize(ISubscriber<int>subscriber){vard1=subscriber.Subscribe(_ =>{});vard2=subscriber.Subscribe(_ =>{});vard3=subscriber.Subscribe(_ =>{});// static DisposableBag: DisposableBag.Create(1~7(optimized) or N);disposable=DisposableBag.Create(d1,d2,d3);}voidClose(){// dispose all subscriptiondisposable?.Dispose();}

IDisposabledisposable;voidOnInitialize(ISubscriber<int>subscriber){// use builder pattern, you can use subscription.AddTo(bag)varbag=DisposableBag.CreateBuilder();subscriber.Subscribe(_ =>{}).AddTo(bag);subscriber.Subscribe(_ =>{}).AddTo(bag);subscriber.Subscribe(_ =>{}).AddTo(bag);disposable=bag.Build();// create final composite IDisposable}voidClose(){// dispose all subscriptiondisposable?.Dispose();}

IDisposabledisposable;voidOnInitialize(ISubscriber<int>subscriber){varbag=DisposableBag.CreateBuilder();// calling once(or x count), you can use DisposableBag.CreateSingleAssignment to hold subscription reference.vard=DisposableBag.CreateSingleAssignment();// you can invoke Dispose in handler action.// assign disposable, you can use `SetTo` and `AddTo` bag.// or you can use d.Disposable = subscriber.Subscribe();subscriber.Subscribe(_ =>{d.Dispose();}).SetTo(d).AddTo(bag);disposable=bag.Build();}voidClose(){disposable?.Dispose();}

The returned IDisposable value must be handled. If it is ignored, it will leak. However Weak reference, which is widely used in WPF, is an anti-pattern. All subscriptions should be managed explicitly.

You can monitor subscription count by MessagePipeDiagnosticsInfo. It can get from service provider(or DI).

publicsealedclassMessagePipeDiagnosticsInfo{/// <summary>Get current subscribed count.</summary>publicintSubscribeCount{get;}/// <summary>/// When MessagePipeOptions.EnableCaptureStackTrace is enabled, list all stacktrace on subscribe./// </summary>publicStackTraceInfo[]GetCapturedStackTraces(boolascending=true);/// <summary>/// When MessagePipeOptions.EnableCaptureStackTrace is enabled, groped by caller of subscribe./// </summary>publicILookup<string,StackTraceInfo>GetGroupedByCaller(boolascending=true)}

If you monitor SubscribeCount, you can check leak of subscription.

publicclassMonitorTimer:IDisposable{CancellationTokenSourcects=newCancellationTokenSource();publicMonitorTimer(MessagePipeDiagnosticsInfodiagnosticsInfo){RunTimer(diagnosticsInfo);}asyncvoidRunTimer(MessagePipeDiagnosticsInfodiagnosticsInfo){while(!cts.IsCancellationRequested){// show SubscribeCountConsole.WriteLine("SubscribeCount:"+diagnosticsInfo.SubscribeCount);awaitTask.Delay(TimeSpan.FromSeconds(5),cts.Token);}}publicvoidDispose(){cts.Cancel();}}

Also, by enabling MessagePipeOptions.EnableCaptureStackTrace (disabled by default), the location of the subscribed location can be displayed, making it easier to find the location of the leak if it exists.

Check the Count of GroupedByCaller, and if any of them show abnormal values, then the stack trace is where it occurs, and you probably ignore Subscription.

for Unity, Window -> MessagePipe Diagnostics window is useful for monitoring subscritpion. It visualizes MessagePipeDianogsticsInfo.

To Enable use of the MessagePipeDiagnostics window, require to set up GlobalMessagePipe.

// VContainerpublicclassMessagePipeDemo:VContainer.Unity.IStartable{publicMessagePipeDemo(IObjectResolverresolver){// require this line.GlobalMessagePipe.SetProvider(resolver.AsServiceProvider());}}// ZenjectvoidConfigure(DiContainercontainer){GlobalMessagePipe.SetProvider(container.AsServiceProvider());}// builtinvarprodiver=builder.BuildServiceProvider();GlobalMessagePipe.SetProvider(provider);

Analyzer

In previous section, we anounce The returned IDisposable value **must** be handled. To prevent subscription leak, we provide roslyn analyzer.

PM> Install-Package MessagePipe.Analyzer

This will raise an error for unhandled Subscribe.

This analyzer can use after Unity 2020.2(see: Roslyn analyzers and ruleset files document). MessagePipe.Analyzer.dll exists in releases page.

Currently Unity's analyzer support is incomplete. We are complementing analyzer support with editor extension, please check the Cysharp/CsprojModifier.

IDistributedPubSub / MessagePipe.Redis

For the distributed(networked) Pub/Sub, you can use IDistributedPublisher<TKey, TMessage>, IDistributedSubscriber<TKey, TMessage> instead of IAsyncPublisher.

publicinterfaceIDistributedPublisher<TKey,TMessage>{ValueTaskPublishAsync(TKeykey,TMessagemessage,CancellationTokencancellationToken=default);}publicinterfaceIDistributedSubscriber<TKey,TMessage>{// and also without filter overload.publicValueTask<IAsyncDisposable>SubscribeAsync(TKeykey,IMessageHandler<TMessage>handler,MessageHandlerFilter<TMessage>[]filters,CancellationTokencancellationToken=default);publicValueTask<IAsyncDisposable>SubscribeAsync(TKeykey,IAsyncMessageHandler<TMessage>handler,AsyncMessageHandlerFilter<TMessage>[]filters,CancellationTokencancellationToken=default);}

IAsyncPublisher means in-memory Pub/Sub. Since processing over the network is fundamentally different, you need to use a different interface to avoid confusion.

Redis is available as a standard network provider.

PM> Install-Package MessagePipe.Redis

use AddMessagePipeRedis to enable redis provider.

Host.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{services.AddMessagePipe().AddRedis(IConnectionMultiplexer|IConnectionMultiplexerFactory,configure);})

IConnectionMultiplexer overload, you can pass StackExchange.Redis's ConnectionMultiplexer directly. Implement own IConnectionMultiplexerFactory to allow for per-key distribution and use from connection pools.

MessagePipeRedisOptions, you can configure serialization.

publicsealedclassMessagePipeRedisOptions{publicIRedisSerializerRedisSerializer{get;set;}}publicinterfaceIRedisSerializer{byte[]Serialize<T>(Tvalue);TDeserialize<T>(byte[]value);}

In default uses MessagePack for C#'s ContractlessStandardResolver. You can change to use other MessagePackSerializerOptions by new MessagePackRedisSerializer(options) or implement own serializer wrapper.

MessagePipe has in-memory IDistributedPublisher/Subscriber for local test usage.

Host.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{varconfig=ctx.Configuration.Get<MyConfig>();varbuilder=services.AddMessagePipe();if(config.IsLocal){// use in-memory IDistributedPublisher/Subscriber in local.builder.AddInMemoryDistributedMessageBroker();}else{// use Redis IDistributedPublisher/Subscriberbuilder.AddRedis();}});

InterprocessPubSub, IRemoteAsyncRequest / MessagePipe.Interprocess

For the interprocess(NamedPipe/UDP/TCP) Pub/Sub(IPC), you can use IDistributedPublisher<TKey, TMessage>, IDistributedSubscriber<TKey, TMessage> similar as MessagePipe.Redis.

PM> Install-Package MessagePipe.Interprocess

MessagePipe.Interprocess is also exsits on Unity(except NamedPipe).

use AddUdpInterprocess, AddTcpInterprocess, AddNamedPipeInterprocess, AddUdpInterprocessUds, AddTcpInterprocessUds to enable interprocess provider(Uds is Unix domain socket, most performant option).

Host.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{services.AddMessagePipe().AddUdpInterprocess("127.0.0.1",3215,configure);// setup host and port.// .AddTcpInterprocess("127.0.0.1", 3215, configure);// .AddNamedPipeInterprocess("messagepipe-namedpipe", configure);// .AddUdpInterprocessUds("domainSocketPath")// .AddTcpInterprocessUds("domainSocketPath")})

publicasyncP(IDistributedPublisher<string,int> publisher){// publish value to remote process.awaitpublisher.PublishAsync("foobar",100);}publicasyncS(IDistributedSubscriber<string,int> subscriber){// subscribe remote-message with "foobar" key.awaitsubscriber.SubscribeAsync("foobar", x =>{Console.WriteLine(x);});}

when injected IDistributedPublisher, process will be server, start to listen client. when injected IDistributedSubscriber, process will be client, start to connect to server. when DI scope is closed, server/client connection is closed.

Udp is connectionless protocol so does not require server is started before client connect. However protocol limitation, does not send over 64K message. We're recommend to use this if message is not large.

Namedpipe is 1:1 connection, can not connect multiple subscribers.

Tcp has no such restrictions and is the most flexible of all the options.

In default uses MessagePack for C#'s ContractlessStandardResolver for message serialization. You can change to use other MessagePackSerializerOptions by MessagePipeInterprocessOptions.MessagePackSerializerOptions.

builder.AddUdpInterprocess("127.0.0.1",3215, options =>{// You can configure other options, `InstanceLifetime` and `UnhandledErrorHandler`.options.MessagePackSerializerOptions=StandardResolver.Options;});

For IPC-RPC, you can use IRemoteRequestHandler<in TRequest, TResponse> that invoke remote IAsyncRequestHandler<TRequest, TResponse>. using TcpInterprocess or NamedPipeInterprocess enabled it.

Host.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{services.AddMessagePipe().AddTcpInterprocess("127.0.0.1",3215, x =>{x.HostAsServer=true;// if remote process as server, set true(otherwise false(default)).});});

// example: server handlerpublicclassMyAsyncHandler:IAsyncRequestHandler<int,string>{publicasyncValueTask<string>InvokeAsync(intrequest,CancellationTokencancellationToken=default){awaitTask.Delay(1);if(request==-1){thrownewException("NO -1");}else{return"ECHO:"+request.ToString();}}}

// clientasyncvoidA(IRemoteRequestHandler<int,string>remoteHandler){varv=awaitremoteHandler.InvokeAsync(9999);Console.WriteLine(v);// ECHO:9999}

For Unity, requires to import MessagePack-CSharp package and needs slightly different configuration.

// example of VContainervarbuilder=newContainerBuilder();varoptions=builder.RegisterMessagePipe(configure);varmessagePipeBuilder=builder.ToMessagePipeBuilder();// require to convert ServiceCollection to enable IntereprocessvarinterprocessOptions=messagePipeBuilder.AddTcpInterprocess();// register manually.// IDistributedPublisher/SubscribermessagePipeBuilder.RegisterTcpInterprocessMessageBroker<int,int>(interprocessOptions);// RemoteHandlerbuilder.RegisterAsyncRequestHandler<int,string,MyAsyncHandler>(options);// for servermessagePipeBuilder.RegisterTcpRemoteRequestHandler<int,string>(interprocessOptions);// for client

MessagePipeOptions

You can configure MessagePipe behaviour by MessagePipeOptions in AddMessagePipe(Action<MMessagePipeOptions> configure).

Host.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{// var config = ctx.Configuration.Get<MyConfig>(); // optional: get settings from configuration(use it for options configure)services.AddMessagePipe(options =>{options.InstanceLifetime=InstanceLifetime.Scoped;
#if DEBUG// EnableCaptureStackTrace slows performance, so recommended to use only in DEBUG and in profiling, disable it.options.EnableCaptureStackTrace=true;
#endif
});})

Option has these properties(and method).

publicsealedclassMessagePipeOptions{AsyncPublishStrategyDefaultAsyncPublishStrategy;// default is ParallelHandlingSubscribeDisposedPolicyHandlingSubscribeDisposedPolic;// default is IgnoreInstanceLifetimeInstanceLifetime;// default is SingletonInstanceLifetimeRequestHandlerLifetime;// default is ScopedboolEnableAutoRegistration;// default is trueboolEnableCaptureStackTrace;// default is falsevoidSetAutoRegistrationSearchAssemblies(paramsAssembly[]assemblies);voidSetAutoRegistrationSearchTypes(paramsType[]types);voidAddGlobal***Filter<T>();}publicenumAsyncPublishStrategy{Parallel,Sequential}publicenumInstanceLifetime{Singleton,Scoped,Transient}publicenumHandlingSubscribeDisposedPolicy{Ignore,Throw}

DefaultAsyncPublishStrategy

IAsyncPublisher has PublishAsync method. If AsyncPublishStrategy.Sequential, await each subscribers. If Parallel, uses WhenAll.

publicinterfaceIAsyncPublisher<TMessage>{// using Default AsyncPublishStrategyValueTaskPublishAsync(TMessagemessage,CancellationTokencancellationToken=default);ValueTaskPublishAsync(TMessagemessage,AsyncPublishStrategypublishStrategy,CancellationTokencancellationToken=default);// snip others...}publicinterfaceIAsyncPublisher<TKey,TMessage>whereTKey:notnull{// using Default AsyncPublishStrategyValueTaskPublishAsync(TKeykey,TMessagemessage,CancellationTokencancellationToken=default);ValueTaskPublishAsync(TKeykey,TMessagemessage,AsyncPublishStrategypublishStrategy,CancellationTokencancellationToken=default);// snip others...}publicinterfaceIAsyncRequestAllHandler<inTRequest,TResponse>{// using Default AsyncPublishStrategyValueTask<TResponse[]>InvokeAllAsync(TRequestrequest,CancellationTokencancellationToken=default);ValueTask<TResponse[]>InvokeAllAsync(TRequestrequest,AsyncPublishStrategypublishStrategy,CancellationTokencancellationToken=default);// snip others...}

MessagePipeOptions.DefaultAsyncPublishStrategy's default is Parallel.

HandlingSubscribeDisposedPolicy

When ISubscriber.Subscribe after MessageBroker(publisher/subscriber manager) is disposed(for example, scope is disposed), choose Ignore(returns empty IDisposable) or Throw exception. Default is Ignore.

InstanceLifetime

Configure MessageBroker(publisher/subscriber manager)'s lifetime of DI cotainer. You can choose Singleton or Scoped. Default is Singleton. When choose Scoped, each messagebrokers manage different subscribers and when scope is disposed, unsubscribe all managing subscribers.

RequestHandlerLifetime

Configure IRequestHandler/IAsyncRequestHandler's lifetime of DI container. You can choose Singleton or Scoped or Transient. Default is Scoped.

EnableAutoRegistration/SetAutoRegistrationSearchAssemblies/SetAutoRegistrationSearchTypes

Register IRequestHandler, IAsyncHandler and filters to DI container automatically on startup. Default is true and default search target is CurrentDomain's all assemblies and types. However, this sometimes fails to detect the assembly being stripped. In that case, you can enable the search by explicitly adding it to SetAutoRegistrationSearchAssemblies or SetAutoRegistrationSearchTypes.

[IgnoreAutoRegistration] attribute allows to disable auto registration which attribute attached.

EnableCaptureStackTrace

See the details Managing Subscription and Diagnostics section, if true then capture stacktrace on Subscribe. It is useful for debugging but performance will be degraded. Default is false and recommended to enable only debug.

AddGlobal***Filter

Add global filter, for example logging filter will be useful.

publicclassLoggingFilter<T>:MessageHandlerFilter<T>{readonlyILogger<LoggingFilter<T>>logger;publicLoggingFilter(ILogger<LoggingFilter<T>>logger){this.logger=logger;}publicoverridevoidHandle(Tmessage,Action<T>next){try{logger.LogDebug("before invoke.");next(message);logger.LogDebug("invoke completed.");}catch(Exceptionex){logger.LogError(ex,"error");}}}

To enable all types, use open generics.

Host.CreateDefaultBuilder().ConfigureServices((ctx,services)=>{services.AddMessagePipe(options =>{// use typeof(Filter<>, order);options.AddGlobalMessageHandlerFilter(typeof(LoggingFilter<>),-10000);});});

Global provider

If you want to get publisher/subscriber/handler from globally scope, get IServiceProvider before run and set to static helper called GlobalMessagePipe.

varhost=Host.CreateDefaultBuilder().ConfigureServices((ctx,x)=>{x.AddMessagePipe();}).Build();// build host before run.GlobalMessagePipe.SetProvider(host.Services);// set service providerawaithost.RunAsync();// run framework.

GlobalMessagePipe has these static method(GetPublisher<T>, GetSubscriber<T>, CreateEvent<T>, etc...) so you can get globally.

Integration with other DI library

All(popular) DI libraries has Microsoft.Extensions.DependencyInjection bridge so configure by MS.E.DI and use bridge if you want.

Compare with Channels

System.Threading.Channels(for Unity, UniTask.Channels) uses Queue internal, the producer is not affected by the performance of the consumer, and the consumer can control the flow rate(back pressure). This is a different use than MessagePipe's Pub/Sub.

Unity

You need to install Core library and choose VContainer or Zenject or BuiltinContainerBuilder for runtime. You can install via UPM git URL package or asset package(MessagePipe.*.unitypackage) available in MessagePipe/releases page.

Core https://github.com/Cysharp/MessagePipe.git?path=src/MessagePipe.Unity/Assets/Plugins/MessagePipe
VContainer https://github.com/Cysharp/MessagePipe.git?path=src/MessagePipe.Unity/Assets/Plugins/MessagePipe.VContainer
Zenject https://github.com/Cysharp/MessagePipe.git?path=src/MessagePipe.Unity/Assets/Plugins/MessagePipe.Zenject

Andalso, requires UniTask to install, all ValueTask declaration in .NET is replaced to UniTask.

Note

Unity version does not have open generics support(for IL2CPP) and does not support auto registration. Therefore, all required types need to be manually registered.

VContainer's installation sample.

publicclassGameLifetimeScope:LifetimeScope{protectedoverridevoidConfigure(IContainerBuilderbuilder){// RegisterMessagePipe returns options.varoptions=builder.RegisterMessagePipe(/* configure option */);// Setup GlobalMessagePipe to enable diagnostics window and global functionbuilder.RegisterBuildCallback(c =>GlobalMessagePipe.SetProvider(c.AsServiceProvider()));// RegisterMessageBroker: Register for IPublisher<T>/ISubscriber<T>, includes async and buffered.builder.RegisterMessageBroker<int>(options);// also exists RegisterMessageBroker<TKey, TMessage>, RegisterRequestHandler, RegisterAsyncRequestHandler// RegisterMessageHandlerFilter: Register for filter, also exists RegisterAsyncMessageHandlerFilter, Register(Async)RequestHandlerFilterbuilder.RegisterMessageHandlerFilter<MyFilter<int>>();builder.RegisterEntryPoint<MessagePipeDemo>(Lifetime.Singleton);}}publicclassMessagePipeDemo:VContainer.Unity.IStartable{readonlyIPublisher<int>publisher;readonlyISubscriber<int>subscriber;publicMessagePipeDemo(IPublisher<int>publisher,ISubscriber<int>subscriber){this.publisher=publisher;this.subscriber=subscriber;}publicvoidStart(){vard=DisposableBag.CreateBuilder();subscriber.Subscribe(x =>Debug.Log("S1:"+x)).AddTo(d);subscriber.Subscribe(x =>Debug.Log("S2:"+x)).AddTo(d);publisher.Publish(10);publisher.Publish(20);publisher.Publish(30);vardisposable=d.Build();disposable.Dispose();}}

Tip

If you are using Unity 2022.1 or later and VContainer 1.14.0 or later, you do not need RegsiterMessageBroker<>. A set of types including ISubscriber<>, IPublisher<> or its asynchronous version will be resolved automatically. Note that IRequesthandler<> and IRequestAllHanlder<> still require manual registration.

Unity version does not have open generics support(for IL2CPP) and does not support auto registration. Therefore, all required types need to be manually registered.

Zenject's installation sample.

voidConfigure(DiContainerbuilder){// BindMessagePipe returns options.varoptions=builder.BindMessagePipe(/* configure option */);// BindMessageBroker: Register for IPublisher<T>/ISubscriber<T>, includes async and buffered.builder.BindMessageBroker<int>(options);// also exists BindMessageBroker<TKey, TMessage>, BindRequestHandler, BindAsyncRequestHandler// BindMessageHandlerFilter: Bind for filter, also exists BindAsyncMessageHandlerFilter, Bind(Async)RequestHandlerFilterbuilder.BindMessageHandlerFilter<MyFilter<int>>();// set global to enable diagnostics window and global functionGlobalMessagePipe.SetProvider(builder.AsServiceProvider());}

Zenject version is not supported InstanceScope.Singleton for Zenject's limitation. The default is Scoped, which cannot be changed.

BuiltinContainerBuilder is builtin minimum DI library for MessagePipe, it no needs other DI library to use MessagePipe. Here is installation sample.

varbuilder=newBuiltinContainerBuilder();builder.AddMessagePipe(/* configure option */);// AddMessageBroker: Register for IPublisher<T>/ISubscriber<T>, includes async and buffered.builder.AddMessageBroker<int>(options);// also exists AddMessageBroker<TKey, TMessage>, AddRequestHandler, AddAsyncRequestHandler// AddMessageHandlerFilter: Register for filter, also exists RegisterAsyncMessageHandlerFilter, Register(Async)RequestHandlerFilterbuilder.AddMessageHandlerFilter<MyFilter<int>>();// create provider and set to Global(to enable diagnostics window and global fucntion)varprovider=builder.BuildServiceProvider();GlobalMessagePipe.SetProvider(provider);// --- to use MessagePipe, you can use from GlobalMessagePipe.varp=GlobalMessagePipe.GetPublisher<int>();vars=GlobalMessagePipe.GetSubscriber<int>();vard=s.Subscribe(x =>Debug.Log(x));p.Publish(10);p.Publish(20);p.Publish(30);d.Dispose();

BuiltinContainerBuilder does not supports scope(always InstanceScope.Singleton), IRequestAllHandler/IAsyncRequestAllHandler, and many DI functionally, so we recommend to use by GlobalMessagePipe when use BuiltinContainerBuilder.

Adding global filter, you can not use open generics filter so recommended to create these helper method.

// Register IPublisher<T>/ISubscriber<T> and global filter.staticvoidRegisterMessageBroker<T>(IContainerBuilderbuilder,MessagePipeOptionsoptions){builder.RegisterMessageBroker<T>(options);// setup for global filters.options.AddGlobalMessageHandlerFilter<MyMessageHandlerFilter<T>>();}// Register IRequestHandler<TReq, TRes>/IRequestAllHandler<TReq, TRes> and global filter.staticvoidRegisterRequest<TRequest,TResponse,THandler>(IContainerBuilderbuilder,MessagePipeOptionsoptions)whereTHandler:IRequestHandler{builder.RegisterRequestHandler<TRequest,TResponse,THandler>(options);// setup for global filters.options.AddGlobalRequestHandlerFilter<MyRequestHandlerFilter<TRequest,TResponse>>();}

Also you can use GlobalMessagePipe and MessagePipe Diagnostics window. see: Global provider and Managing Subscription and Diagnostics section.

License

This library is licensed under the MIT License.

Name		Name	Last commit message	Last commit date
.github		.github
docs		docs
sandbox		sandbox
src		src
tests		tests
tools/PostBuildUtility		tools/PostBuildUtility
.dockerignore		.dockerignore
.editorconfig		.editorconfig
.gitignore		.gitignore
Directory.Build.props		Directory.Build.props
Icon.png		Icon.png
LICENSE		LICENSE
MessagePipe.sln		MessagePipe.sln
MessagePipe.sln.startup.json		MessagePipe.sln.startup.json
NuGet.config		NuGet.config
README.md		README.md
docker-compose.yml		docker-compose.yml
opensource.snk		opensource.snk

Navigation Menu

Uh oh!

Repository files navigation

MessagePipe

Getting Started

Publish/Subscribe

ISingleton, IScoped

Buffered

EventFactory

Request/Response/All

Subscribe Extensions

Filter

Managing Subscription and Diagnostics

Analyzer

IDistributedPubSub / MessagePipe.Redis

InterprocessPubSub, IRemoteAsyncRequest / MessagePipe.Interprocess

MessagePipeOptions

DefaultAsyncPublishStrategy

HandlingSubscribeDisposedPolicy

InstanceLifetime

RequestHandlerLifetime

EnableAutoRegistration/SetAutoRegistrationSearchAssemblies/SetAutoRegistrationSearchTypes

EnableCaptureStackTrace

AddGlobal***Filter

Global provider

Integration with other DI library

Compare with Channels

Unity

License

About

Releases

Packages

Used by

Contributors

Languages

Uh oh!

Latest commit

History

Folders and files

Repository files navigation

MessagePipe

Getting Started

Publish/Subscribe

ISingleton***, IScoped***

Buffered

EventFactory

Request/Response/All

Subscribe Extensions

Filter

Managing Subscription and Diagnostics

Analyzer

IDistributedPubSub / MessagePipe.Redis

InterprocessPubSub, IRemoteAsyncRequest / MessagePipe.Interprocess

MessagePipeOptions

DefaultAsyncPublishStrategy

HandlingSubscribeDisposedPolicy

InstanceLifetime

RequestHandlerLifetime

EnableAutoRegistration/SetAutoRegistrationSearchAssemblies/SetAutoRegistrationSearchTypes

EnableCaptureStackTrace

AddGlobal***Filter

Global provider

Integration with other DI library

Compare with Channels

Unity

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages

ISingleton, IScoped