Sai.Messaging 1.0.3

Sai.Messaging

Messaging utility library.

Usage

Publisher/Subscriber Channel

var messageService = new InProcessMessageService();
var pubSubChannel = messageService.GetPublisherSubscriberChannel<int>("my-topic"); // Creates or gets a channel for the topic
var reader1 = pubSubChannel.Subscribe();
var reader2 = pubSubChannel.Subscribe();

var writerTask = Task.Run(async () =>
{
    for (int i = 0; i < MESSAGE_COUNT; i++)
    {
        await pubSubChannel.WriteAsync(i);
    }
});

var readerTask1 = Task.Run(async () =>
{
    int value;
    do
    {
        value = await reader1.ReadAsync();
    } while (value < MESSAGE_COUNT - 1);
});

var readerTask2 = Task.Run(async () =>
{
    int value;
    do
    {
        value = await reader2.ReadAsync();
    } while (value < MESSAGE_COUNT - 1);
});

Queue Channel

var messageService = new InProcessMessageService();
var queue = messageService.GetQueueChannel<int>("my-queue"); // Creates or gets a queue by name

var writerTask = Task.Run(async () =>
{
    for (int i = 0; i < MESSAGE_COUNT; i++)
    {
        await queue.WriteAsync(i);
    }
});

var readerTask = Task.Run(async () =>
{
    int value;
    do
    {
        value = await queue.ReadAsync();
    } while (value < MESSAGE_COUNT - 1);
});

General Comments

Channel<T> is used for pub/sub and queue implementations as it is fast (see here: https://michaelscodingspot.com/performance-of-producer-consumer/, plus benchmarks in this project). Blocking implementations may be useful in some scenarios, to avoid async/await - however it was found that using BlockingCollection is significantly slower than channels (again, see benchmarks). ConcurrentQueue<T> is much faster than BlockingCollection<T>, however it has a very limited API: Enqueue(), TryDequeue() and TryPeek(), so adding timeout and cancellation would probably make it equivalent to BlockingCollection<T> anyway.