Zero Messenger

Filters allow you to intercept message processing before and after the handler executes. They follow an async decorator pattern similar to ASP.NET Core middleware. Implement IMessageFilter<T> and use the next delegate to pass the message to the next step in the pipeline.

public class LoggingFilter<T> : IMessageFilter<T>
{
    public async ValueTask InvokeAsync(T message, CancellationToken cancellationToken, Func<T, CancellationToken, ValueTask> next)
    {
        Console.WriteLine("Before");
        await next(message, cancellationToken);
        Console.WriteLine("After");
    }
}

To use Zero Messenger in Unity, you must install the NuGet packages using the NugetForUnity tool.

Requirements

• Unity 2021.3 or later

Installation Steps

1. Install NugetForUnity.
2. In Unity, navigate to NuGet > Manage NuGet Packages.
3. Search for the ZeroMessenger package and install it.

By adding the ZeroMessenger.R3 package, you can convert between Zero Messenger interfaces and R3 Observables. This allows you to use R3 operators on message streams or pipe Observable data back into a publisher.

// Install package: dotnet add package ZeroMessenger.R3

// Convert IMessageSubscriber<T> to Observable<T>
subscriber.ToObservable()
    .Subscribe(x => { });

// Connect Observable<T> to IMessagePublisher<T>
observable.SubscribeToPublish(publisher);

// Connect Observable<T> to IMessagePublisher<T>.PublishAsync()
observable.SubscribeAwaitToPublish(publisher, AwaitOperation.Sequential, AsyncPublishStrategy.Parallel);

To use Zero Messenger with Microsoft.Extensions.DependencyInjection, install the ZeroMessenger.DependencyInjection package. Use services.AddZeroMessenger() to register the library in your IServiceCollection.

# .NET CLI
dotnet add package ZeroMessenger.DependencyInjection

# Package Manager
Install-Package ZeroMessenger.DependencyInjection

Host.CreateDefaultBuilder()
    .ConfigureServices((context, services) =>
    {
        // Add Zero Messenger
        services.AddZeroMessenger();

        services.AddSingleton<ServiceA>();
        services.AddSingleton<ServiceB>();
    })
    .Build()
    .Run();

You can add filters in three ways:

1. Directly to a MessageBroker<T>: Use AddFilter<T>(). Filters are applied in the order they are added.
2. Globally via Dependency Injection: Configure filters within AddZeroMessenger() during service registration.
3. Per-subscription: Use the WithFilter<T>() or WithFilters() extension methods on an IMessageSubscriber<T>.

// 1. To a broker
var broker = new MessageBroker<int>();
broker.AddFilter<LoggingFilter<int>>();

// 2. Globally in DI
services.AddZeroMessenger(messenger =>
{
    messenger.AddFilter<LoggingFilter<Message>>();
    messenger.AddFilter(typeof(LoggingFilter<>));
});

// 3. Per subscription
subscriber
    .WithFilter<LoggingFilter<Message>()}
    .Subscribe(x => { });

You can use MessageBroker<T>.Default for a global Pub/Sub implementation, or instantiate a local MessageBroker<T> to act like an event or an Rx Subject<T>.

using System;
using ZeroMessenger;

// Subscribe to messages
var subscription = MessageBroker<Message>.Default.Subscribe(x =>
{
    Console.WriteLine(x.Text);
});

// Publish a message
MessageBroker<Message>.Default.Publish(new Message("Hello!"));

// Unsubscribe
subscription.Dispose();

// Type used for the message
public record struct Message(string Text) { }

Zero Messenger requires .NET Standard 2.1 or later. You can install it via NuGet using the .NET CLI or the Package Manager.

# .NET CLI
dotnet add package ZeroMessenger

# Package Manager
Install-Package ZeroMessenger

Zero Messenger provides PredicateFilter<T>. If you pass a Predicate<T> to AddFilter<T>() or WithFilter<T>(), a PredicateFilter<T> is automatically created. This allows you to exclude messages that do not meet specific criteria.

public record struct FooMessage(int Value);

IMessageSubscriber<FooMessage> subscriber;

// Exclude values less than 0
subscriber
    .WithFilter(x => x.Value >= 0)
    .Subscribe(x => { });

The IMessageSubscriber<T> interface provides methods to listen for messages. Use Subscribe() for synchronous handlers (returning an IDisposable to unsubscribe) or SubscribeAwait() for asynchronous handlers.

IMessageSubscriber<Message> subscriber;

// Subscribe to messages
var subscription = subscriber.Subscribe(x =>
{
    Console.WriteLine(x.Text);
});

// Unsubscribe
subscription.Dispose();

// Asynchronous processing
var asyncSubscription = subscriber.SubscribeAwait(async (x, ct) =>
{
    await FooAsync(x, ct);
}, AsyncSubscribeStrategy.Sequential);

The IMessagePublisher<T> interface allows you to send messages. You can use Publish() for fire-and-forget operations or PublishAsync() to wait for all subscribers to finish processing.

IMessagePublisher<Message> publisher;

// Publish a message (Fire-and-forget)
publisher.Publish(new Message("Foo!"));

// Publish a message and wait for all subscribers to finish processing
await publisher.PublishAsync(new Message("Bar!"), AsyncPublishStrategy.Parallel, cancellationToken);