Forem: Federico Alterio

Building Modern Realtime .NET Apps: SignalR, R3, AsyncR3, SignalsDotnet, and Redis

Federico Alterio — Mon, 16 Feb 2026 19:13:53 +0000

Source Code

You can find the fully functional app with .NET Aspire — including Redis, Avalonia, three ASP.NET Core load balanced apps, and a bonus SignalR with SignalR backplane (yes, it’s not a typo!) — in the R3Async Playground repository.

Intro

First of all, I’d like to ask a question: which API do you prefer?

void InvokeSomeApi(Request request, Action<Response> onCompleted);
Task<Response> InvokeSomeApiAsync(Request request, CancellationToken cancellationToken);

Of course, you chose the second one.

We have built-in language support with await.
No need to deal with callbacks manually.
A standard, built-in way to support cancellation.
A standard way to control continuation behavior (TaskAwaiter).
Task already solves this problem in a well-established way — we shouldn’t reinvent it every time.

Second question: which one do you prefer?

record Notification;

interface IConsumer
{
    public Task OnNotification(Notification notification, CancellationToken cancellationToken);
}

interface IProducer
{
    void Subscribe(IConsumer client);
    void Unsubscribe(IConsumer client);
}

record Notification;
interface IProducer
{
    IAsyncEnumerable<Notification> GetNotificationStream();
}

I would have no doubts:

We have built-in language support (await foreach).
We have a standard way to handle unsubscription via .WithCancellation(cancellationToken).
We get LINQ support (in .NET 10 and later).
We don’t need to introduce an explicit IConsumer interface.

Final question: So why is the first approach still the default for many .NET developers using SignalR?

I think it often comes down to a lack of familiarity with reactive patterns.

Do you even think about enumerating a list without using an IEnumerable? That would be crazy — we would be reinventing enumerables every time.

So why on earth are we reinventing streams over and over again?

We already have basically three stream patterns in .NET:

IAsyncEnumerable<T>: Pull-based async stream. Works naturally with await foreach.
Observable<T>: Push-based synchronous stream. see R3
AsyncObservable<T>: Push-based asynchronous stream. See R3Async

Create a (Reactive) Realtime Messaging App with SignalR

So, how can we get started? SignalR supports streaming.

All we need to do is create a hub and return a stream, like this:

public class AppHub(IChatService chatService) : Hub
{
    // Returns a stream of messages for the specified ChatRoomId
    public async IAsyncEnumerable<ChatMessage> GetChatMessages(
        ChatRoomId roomId,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await foreach (var message in todo.WithCancellation(cancellationToken))
        {
            yield return message;
        }
    }

    // Joins a room and broadcasts messages sent by the client to other users
    public async Task JoinRoom(
        ChatRoomId roomId,
        string user,
        IAsyncEnumerable<ChatMessage> messages,
        CancellationToken cancellationToken)
    {
        await foreach (var message in messages.WithCancellation(cancellationToken))
        {
            // TODO: broadcast message to other users
        }
    }
}

Here, we’re omitting how the todo stream is implemented — the focus is on how SignalR handles the streaming endpoint.

The CancellationToken is tied to the client connection (or more accurately, to the client-side enumeration and connection).
yield return will push data into the socket automatically.

In this way, we’re essentially abstracting away SignalR — the client just consumes a standard IAsyncEnumerable.

Client Implementation with Avalonia

For the client, we’re using Avalonia.

Our goals:

Share a single SignalR connection across the entire app.
Dynamically create and dispose of the connection as needed.
Keep track of usage to know when to dispose — basically a reference counting system.

A good solution is to use RefCountLazy<T> from R3Async, which lets us lazily create the connection and automatically dispose it when no one is using it anymore.

public sealed class SignalRHubClient
{
    readonly RefCountLazy<HubConnection> _sharedConnection;

    public SignalRHubClient()
    {
       _sharedConnection = new(async token =>
        {
            var connection = new HubConnectionBuilder()
                             .WithUrl("http://localhost:5062/hub")
                             .Build();

            await connection.StartAsync(token);
            _isConnected.Value = true;
            return new()
            {
                Value = connection,
                Disposable = AsyncDisposable.Create(async () =>
                {
                    await connection.DisposeAsync();
                    _isConnected.Value = false; 
                })
            };
        });
    }

    readonly Signal<bool> _isConnected = new(false);
    public ISignal<bool> IsConnected => _isConnected;

    public async IAsyncEnumerable<ChatMessage> GetChatMessages(ChatRoomId roomId, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        await using var hubRef = await _sharedConnection.GetAsync(cancellationToken);
        var connection = hubRef.Value;
        var stream = connection.StreamAsync<ChatMessage>("GetChatMessages", roomId, cancellationToken);
        await foreach (var message in stream)
        {
            yield return message;
        }
    }

    public async ValueTask JoinRoom(ChatRoomId roomId, string user, IAsyncEnumerable<ChatMessage> messages, CancellationToken cancellationToken = default)
    {
        await using var hubRef = await _sharedConnection.GetAsync(cancellationToken);
        await hubRef.Value.InvokeAsync("JoinRoom", roomId, user, messages, cancellationToken: cancellationToken);
    }
}

Basically, RefCountLazy<T> is a thread-safe helper that tracks references to our Hub connection.

GetAsync() returns a "disposable reference" to the connection.
Calling DisposeAsync() on that reference decrements the reference count.
When the reference count reaches 0, the connection is automatically disposed.
When the reference count goes from 0 to 1, the connection is created again — and so on.

Effectively, we are moving the IAsyncEnumerable from the server to the client!

ViewModel Implementation

Here we use R3 and SignalsDotnet to automatically subscribe and unsubscribe to IAsyncEnumerable streams based on property changes.

public class ChatViewModel
{
    readonly SignalRHubClient _hubClient = new();

    public ChatViewModel(IReadOnlySignal<string> userName)
    {
        var computedFactory = ComputedSignalFactory.Default
                                                   .DisconnectEverythingWhen(Disconnected.Values);

        // Command to send messages
        SendCommand = computedFactory
            .ComputedObservable(
                () => _hubClient.IsConnected.Value && !string.IsNullOrWhiteSpace(DraftMessage.Value), 
                () => new(false))
            .ToReactiveCommand(
                _ => ToSendMessage.OnNext(new(userName.Value, DraftMessage.Value))
            );

        // Subscribe to incoming chat messages
        computedFactory.AsyncEffect(async token =>
        {
            await foreach (var message in _hubClient.GetChatMessages(new(RoomName.Value), token))
            {
                AllChatMessages.Add(message);
            }
        }, ConcurrentChangeStrategy.CancelCurrent);

        // Join the chat room and broadcast outgoing messages
        computedFactory.AsyncEffect(async token =>
        {
            await _hubClient.JoinRoom(
                new(RoomName.Value),
                userName.Value,
                ToSendMessage.TakeUntil(token).ToAsyncEnumerable(),
                token
            );
        }, ConcurrentChangeStrategy.CancelCurrent);

        // Clear draft message after sending
        ToSendMessage.Subscribe(_ => DraftMessage.Value = "");
    }

    public Signal<bool> Disconnected { get; } = new(true);
    public Signal<string> DraftMessage { get; } = new();
    public Subject<ChatMessage> ToSendMessage { get; } = new();
    public Signal<string> RoomName { get; } = new("MyRoom");
    public ObservableCollection<ChatMessage> AllChatMessages { get; } = new();
    public ICommand SendCommand { get; }
}

See what’s happening here? We’re basically just enumerating an IAsyncEnumerable and adding items to a list — it really doesn’t get simpler than that.

SignalsDotnet takes care of all the dependency tracking for us:

Whenever RoomName.Value changes, the CancelCurrent strategy cancels the original stream.
This cancellation is automatically propagated to the server.
A new stream is then enumerated for the updated room — all automatically.
Everything is disposed automatically if the view is unloaded

Ok, but how can we implement the backend?

Let’s go step by step.

First assumption: We have a single backend service — every client connects directly to this server.

Later, we can scale out by implementing a Redis backplane (we won’t use the built-in SignalR backplane for this, but we implement it from scratch) to synchronize messages across multiple server instances.

Understand What We Need

So, what do we actually need?

We need a way to publish messages.

Don’t worry about the underlying technology yet — focus only on the method signature and how the API should look.

So we need something like this:

public interface IPublisher<T>
{
    IAsyncDisposable SubscribeAsync(Func<T, CancellationToken, ValueTask> onNextAsync);
    ValueTask PublishAsync(T value, CancellationToken cancellationToken);
}

Here’s the fun part: stop trying to implement this interface yourself.

This problem is already solved and has a well-known name in the ReactiveX world — it’s called a Subject.

All we need is the async version of it, and we can use R3Async to get exactly that.
So, what we really need is an ISubject.

But is this enough? Not quite.

We need a subject per topic, so it’s more like:

  ConcurrentDictionary<string, ISubject<T>> _publishersByTopic;

Are we good now? Still not. We want the publisher to exist only if someone is subscribed and to dispose automatically when no one needs it. We don't want the dictionary to have zombie-references

That means we need reference counting again, but this time based on a key (the topic). We can use RefCountTable for this! It's basically a concurrent dictionary with only GetOrAddAsyncmethod, that counts how many active references there are.
That means we need a

    readonly RefCountTable<ChatRoomId, ISubject<ChatMessage>> _roomPublisherByRoomId;

Maybe we can hide it behind an interface like this:

public interface IChatService
{
    ValueTask<IAsyncDisposableReference<ISubject<ChatMessage>>> GetOrCreateChatRoom(ChatRoomId id, CancellationToken cancellationToken);
}

Here, IAsyncDisposableReference is essentially the ref-counted reference.
Calling DisposeAsync() on it signals that we no longer need the publisher, and when the ref count reaches 0, it will automatically dispose the underlying implementation that exposes the subject.

Completing the hub

We are now ready to complete the Hub implementation:

public class AppHub(IChatService chatService) : Hub
{
    public async IAsyncEnumerable<ChatMessage> GetChatMessages(ChatRoomId roomId, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        await using var chatRef = await chatService.GetOrCreateChatRoom(roomId, cancellationToken);
        var chatMessages = chatRef.Value;
        await foreach (var message in chatMessages.Values
                                          .ToAsyncEnumerable(static () => Channel.CreateUnbounded<ChatMessage>())
                                          .WithCancellation(cancellationToken))
        {
            yield return message;
        }
    }

    public async Task JoinRoom(ChatRoomId roomId, string user, IAsyncEnumerable<ChatMessage> messages)
    {
        var cancellationToken = Context.ConnectionAborted;
        await using var chatRef = await chatService.GetOrCreateChatRoom(roomId, cancellationToken);
        var chatMessages = chatRef.Value;
        await chatMessages.OnNextAsync(new ChatMessage(user, $"{user} joined the room"), cancellationToken);
        try
        {
            await foreach (var message in messages.WithCancellation(cancellationToken))
            {
                await chatMessages.OnNextAsync(message, cancellationToken);
            }
        }
        finally
        {
            await chatMessages.OnNextAsync(new ChatMessage(user, $"{user} left the room"), cancellationToken);
        }
    }
}

How it works

JoinRoom will return only when the user has left the room.
Every message from the client is published to the subject.
GetChatMessages enumerates every notification from that subject and yields it to the client in real time.
References from RefCountTable are easy to manage using await using statements. They automatically track the lifetime of both publishers and consumers.
Note that we can move all this method outside the hub class, since we are not depending from SignalR anywhere. That means that we can easily unit test the logic also

Ok, but how is `IChatService` implemented?

Here’s the first implementation: InMemory

public class InMemoryChatService : IChatService
{
    readonly RefCountTable<ChatRoomId, ISubject<ChatMessage>> _chatByRoom;

    public InMemoryChatService()
    {
        _chatByRoom = new(async (roomId, cancellationToken) =>
        {
            var chat = Subject.Create<ChatMessage>(new SubjectCreationOptions
            {
                IsStateless = false,
                PublishingOption = PublishingOption.Concurrent
            });

            return new()
            {
                Value = chat,
                Disposable = AsyncDisposable.Create(() => chat.OnCompletedAsync(Result.Success))
            };
        });
    }

    public ValueTask<IAsyncDisposableReference<ISubject<ChatMessage>>> GetOrCreateChatRoom(ChatRoomId id, CancellationToken cancellationToken)
    {
        return _chatByRoom.GetOrCreateAsync(id, cancellationToken);
    }
}

Ok, this is good and all, but how do you scale it out?

Just give another implementation of the subject (publisher)!
Let's use redis as backplane

sealed class RedisSubscriberSubject<T> : ISubject<T> where T : class
{
    readonly ISubscriber _subscriber;
    readonly RedisChannel _channel;

    public RedisSubscriberSubject(ISubscriber subscriber, 
                                  RedisChannel channel)
    {
        _subscriber = subscriber;
        _channel = channel;

        Values = AsyncObservable.Create<T>((observer, _) =>
        {
            var messageQueue = subscriber.Subscribe(_channel);
            messageQueue.OnMessage(async message =>
            {
                var notification = JsonSerializer.Deserialize<Notification>((string)message.Message!)!;
                await notification.ForwardTo(observer, CancellationToken.None);
            });

            var disposable = AsyncDisposable.Create(async () => await messageQueue.UnsubscribeAsync());
            return new(disposable);
        });
    }

    public AsyncObservable<T> Values { get; }

    public ValueTask OnNextAsync(T value, CancellationToken cancellationToken) => ForwardNotification(Notification.FromOnNext(value));
    public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken) => ForwardNotification(Notification.FromOnErrorResume(error));
    public ValueTask OnCompletedAsync(Result result) => ForwardNotification(Notification.FromOnCompleted(result));
    async ValueTask ForwardNotification(Notification notification)
    {
        var json = JsonSerializer.Serialize(notification);
        await _subscriber.PublishAsync(_channel, json);
    }

    record Notification
    {
        public static Notification FromOnNext(T value) => new()
        {
            Value = value,
            ErrorMessage = null,
            IsCompleted = false
        };

        public static Notification FromOnErrorResume(Exception exception) => new()
        {
            Value = null,
            ErrorMessage = exception.Message,
            IsCompleted = false
        };

        public static Notification FromOnCompleted(Result result) => new()
        {
            Value = null,
            ErrorMessage = result.Exception?.Message,
            IsCompleted = true
        };

        public string? ErrorMessage { get; init; }
        public T? Value { get; init; }
        public bool IsCompleted { get; init; }

        public ValueTask ForwardTo(AsyncObserver<T> observer, CancellationToken cancellationToken) => (IsCompleted, ErrorMessage) switch
        {
            (false, null) => observer.OnNextAsync(Value!, cancellationToken),
            (false, not null) => observer.OnErrorResumeAsync(new(ErrorMessage), cancellationToken),
            (true, null) => observer.OnCompletedAsync(Result.Success),
            (true, not null) => observer.OnCompletedAsync(Result.Failure(new(ErrorMessage)))
        };
    }
}

public static class RedisSubscriberEx
{
    public static ISubject<T> ToSubject<T>(this ISubscriber subscriber, RedisChannel channel) where T : class => new RedisSubscriberSubject<T>(subscriber, channel);
}

the Important bits are

AsyncObservable.Create<T>((observer, _) =>
        {
            var messageQueue = subscriber.Subscribe(_channel);
            messageQueue.OnMessage(async message =>
            {
                var notification = JsonSerializer.Deserialize<Notification>((string)message.Message!)!;
                await notification.ForwardTo(observer, CancellationToken.None);
            });

            var disposable = AsyncDisposable.Create(async () => await messageQueue.UnsubscribeAsync());
            return new(disposable);
        });

Here we are mapping the redis subscription to an AsyncObservable.

For publishing to Redis:

    public ValueTask OnNextAsync(T value, CancellationToken cancellationToken) => ForwardNotification(Notification.FromOnNext(value));
    public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken) => ForwardNotification(Notification.FromOnErrorResume(error));
    public ValueTask OnCompletedAsync(Result result) => ForwardNotification(Notification.FromOnCompleted(result));
    async ValueTask ForwardNotification(Notification notification)
    {
        var json = JsonSerializer.Serialize(notification);
        await _subscriber.PublishAsync(_channel, json);
    }

Now we can implement our RedisChatService like this:

public class RedisChatService : IChatService
{
    readonly RefCountTable<ChatRoomId, ISubject<ChatMessage>> _table;

    public RedisChatService(IConnectionMultiplexer connectionMultiplexer)
    {
        _table = new(async (roomId, _) =>
        {
            var subscriber = connectionMultiplexer.GetSubscriber();
            var disposable = AsyncDisposable.Create(async () => await subscriber.UnsubscribeAllAsync());
            try
            {
                var subject = subscriber.ToSubject<ChatMessage>(new(roomId.Name, RedisChannel.PatternMode.Auto));
                return new AsyncDisposableValue<ISubject<ChatMessage>>
                {
                    Value = subject,
                    Disposable = disposable
                };
            }
            catch
            {
                await disposable.DisposeAsync();
                throw;
            }
        });
    }

    public ValueTask<IAsyncDisposableReference<ISubject<ChatMessage>>> GetOrCreateChatRoom(ChatRoomId id, CancellationToken cancellationToken)
    {
        return _table.GetOrCreateAsync(id, cancellationToken);
    }
}

As you can see, if we combine the right tools, we can achieve great things with very little effort.

AsyncR3 & RabbitMQ

Federico Alterio — Thu, 22 Jan 2026 19:24:21 +0000

Before reading, check this to know what R3Async is.

In that post, I wrote:

This implementation is better suited to common backend scenarios where a subscription wraps an external service (for example, a remote queue) rather than a local, in-process event source. In such cases, subscribing and unsubscribing must be genuinely asynchronous.

Makes sense, but without a concrete example, it’s just abstract philosophy. Let’s see how this works in practice with RabbitMQ.

Step 1: Add the RabbitMQ.Client Package and R3Async

First, we need to reference the official RabbitMQ .NET client package and
R3Async.

dotnet add package RabbitMQ.Client
dotnet add package R3Async

Step 2: Create a connection to RabbitMQ Server

Connections are expensive, so we want a single connection for the app. It should be established only when needed and disposed when unused.

R3Async helps here:

extension(IConnection)
{
    public static AsyncObservable<IConnection> CreateSharedConnectionObservable(ConnectionFactory connectionFactory)
    {
        return AsyncObservable.Using(
                   async token => await connectionFactory.CreateConnectionAsync(token),
                   static connection => AsyncObservable.CreateAsBackgroundJob<IConnection>(
                       (observer, ct) => observer.OnNextAsync(connection, ct),
                       true))
               .OnErrorResumeAsFailure()
               .StatelessReplayLatestPublish()
               .RefCount();
    }
}

We use:

OnErrorResumeAsFailure() because an error here is fatal.
StatelessReplayLatestPublish() to share the connection (stateless so we don’t remember past errors).
RefCount() to create the connection when the first subscriber appears and dispose it when the last one leaves.

Step 3: Create Channels

Channels in RabbitMQ are more lightweight than connections, but still we can't abuse them. Also they are not thread-safe. Sharing them across threads without protection is risky.

To handle this safely, we can wrap a channel in a small synchronized wrapper that uses a semaphore to control access:

public class SynchronizedChannel(IChannel channel) : IAsyncDisposable
{
    readonly SemaphoreSlim _gate = new(1, 1);
    readonly IChannel _channel = channel;

    public async ValueTask<ChannelReleaser> GetChannelLockAsync(CancellationToken cancellationToken)
    {
        await _gate.WaitAsync(cancellationToken);
        return new ChannelReleaser(this);
    }

    public async ValueTask DisposeAsync()
    {     
     await _channel.DisposeAsync();
    }

    public readonly struct ChannelReleaser(SynchronizedChannel parent) : IDisposable
    {
        public void Dispose()
        {           parent._gate.Release();
        }

        public IChannel Channel => parent._channel;
    }
}

Step 4: Create Shared Channels

Now we are ready to create shared channels.

We can have multiple channels, but the same rules as connections apply:

If no one is using a channel, it should be disposed.
If someone subscribes, it should be created on demand.

This allows to not bloat with channels while keeping access thread-safe with the synchronized wrapper. I'ts a trade-off. We can still create multiple SynchronizedChannel instances if we want to.

public static AsyncObservable<SynchronizedChannel> SharedSynchronizedChannel(
    this IConnection connection,
    Func<IConnection, CancellationToken, ValueTask<IChannel>> channelFactory)
{
    return AsyncObservable.Using(
               async ct =>
               {
                   var channel = await channelFactory(connection, ct);
                   return channel.ToSynchronizedChannel();
               },
               static channel => AsyncObservable.CreateAsBackgroundJob<SynchronizedChannel>(
                   (observer, ct) => observer.OnNextAsync(channel, ct),
                   true))
           .OnErrorResumeAsFailure()
           .StatelessReplayLatestPublish()
           .RefCount();
}

Small note: We use CreateAsBackgroundJob instead of Create so that SubscribeAsync can return even if the channel instance is not completely notified to the observer (SubscribeAsync will still return only after the Channel is created, that would be impossible with sync R3 or Rx.NET).

Step 5: Convert Basic Consume to an AsyncObservable

Now that we have shared, synchronized channels, we can wrap RabbitMQ’s basic consume in an AsyncObservable.

public record BasicConsumerArgs<T>(SynchronizedChannel SyncChannel, T Value, BasicDeliverEventArgs OriginalArgs);

extension(SynchronizedChannel syncChannel)
{
    public AsyncObservable<BasicConsumerArgs<T>> CreateBasicConsumeObservable(
        Func<IChannel, IAsyncBasicConsumer, CancellationToken, ValueTask<string>> basicConsume,
        bool unsubscriptionNoWait = false)
    {
        var observable = AsyncObservable.Create<BasicConsumerArgs<Unit>>(async (observer, subscribeToken) =>
        {
            using var channelLock = await syncChannel.GetChannelLockAsync(subscribeToken);
            var channel = channelLock.Channel;
            var consumer = new AsyncEventingBasicConsumer(channel);
            var cts = new CancellationTokenSource();
            var streamToken = cts.Token;

            consumer.ReceivedAsync += OnReceivedAsync;

            var consumerTag = await basicConsume(channel, consumer, subscribeToken);

            async Task OnReceivedAsync(object sender, BasicDeliverEventArgs args)
            {
                try
                {
                    await observer.OnNextAsync(new(syncChannel, Unit.Default, args), streamToken)
                                  .AsTask()
                                  .WaitAsync(Timeout.InfiniteTimeSpan, streamToken);
                }
                catch (OperationCanceledException)
                {
                }
            }

            return AsyncDisposable.Create(async () =>
            {
                using (cts)
                {
                    consumer.ReceivedAsync -= OnReceivedAsync;
                    await Task.Run(cts.Cancel, CancellationToken.None);
                    using var innerChannelLock = await syncChannel.GetChannelLockAsync(CancellationToken.None);
                    await innerChannelLock.Channel.BasicCancelAsync(consumerTag, unsubscriptionNoWait, CancellationToken.None);
                }
            });
        });

        return observable;
    }       
}

Here, we map the SubscribeAsync of the AsyncObservable to RabbitMQ’s += ReceivedAsync + BasicConsume, the DisposeAsync of the subscription to -= and BasicCancelAsync, and ReceivedAsync to OnNextAsync.

A small detail for clever readers: we handle reentrancy because ReceivedAsync might trigger DisposeAsync while processing a message.

The RabbitMQ client I think does not support disposing a channel from within ReceivedAsync, so which could otherwise lead to deadlocks.

Full Example

We register the shared connection in the DI

builder.Services.AddSingleton<AsyncObservable<IConnection>>(s =>
{
    var uri = s.GetRequiredService<IConfiguration>()["MESSAGING_URI"]!;
    return IConnection.CreateSharedConnectionObservable(new ConnectionFactory
    {
        Uri = new(uri)
    });
});

We create a background service to consume messages (30 maximum messages concurrently):


public class ConsumerHostedService(AsyncObservable<IConnection> sharedConnection) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await sharedConnection.Select(connection => connection.SharedSynchronizedChannel(async (connection, token) => await connection.CreateChannelAsync(cancellationToken: token)))
                              .Switch()
                              .Select(syncChannel => syncChannel.CreateBasicConsumeObservable(async (channel, consumer, token) =>
                              {
                                  await channel.QueueDeclareAsync(queue: "hello", durable: false, exclusive: false, autoDelete: false,
                                                                  arguments: null, cancellationToken: token);

                                  return await channel.BasicConsumeAsync("hello", autoAck: false, consumer: consumer, cancellationToken: token);
                              }))
                              .Switch()
                              .Select(message => AsyncObservable.FromAsync(async token =>
                              {
                                  await message.HandleAndAck(async t =>
                                  {
                                    var body = Encoding.UTF8.GetString(message.OriginalArgs.Body.Span);
                                      Console.WriteLine($" [x] Processing {body}");
                                      await Task.Delay(2000, t);
                                      Console.WriteLine($" [x] Processed {body}");
                                  }, cancellationToken: token);
                              }))
                              .Merge(30)
                              .OnDispose(() => Console.WriteLine("Consumer disposed"))
                              .WaitCompletionAsync(stoppingToken);
    }
}

HandleAsync is just an helper extension methods that acks if function executed succesfully, or nacks otherwise:

Other approaches

Some operators of AsyncObservable are a bit dangerous to use in this context, since we should ack (or nack) messages. We can easly fix this with a thin wrapper

Step 6: Wrap AsyncObservable into a dedicated Akc-aware observable

We can define a small wrapper, BasicConsumerObservable, that wraps AsyncObservable<BasicConsumerArgs<T>> and provides LINQ-style operations safely:

public record struct BasicConsumerObservable<T>(AsyncObservable<BasicConsumerArgs<T>> RawObservable);

// LINQ-like extensions for BasicConsumerObservable
public static class BasicConsumerObservable
{
    public static BasicConsumerObservable<T> ToBasicConsumerObservable<T>(
        this AsyncObservable<BasicConsumerArgs<T>> @this) => new(@this);

    extension<T>(BasicConsumerObservable<T> @this)
    {
        public BasicConsumerObservable<TDest> Select<TDest>(
            Func<BasicConsumerArgs<T>, CancellationToken, ValueTask<TDest>> selector)
        {
            var mappedObservable = @this.RawObservable.Select(async (args, token) =>
            {
                var result = await selector(args, token);
                return new BasicConsumerArgs<TDest>(args.SyncChannel, result, args.OriginalArgs);
            });
            return new BasicConsumerObservable<TDest>(mappedObservable);
        }

        public BasicConsumerObservable<TDest> Select<TDest>(
            Func<BasicConsumerArgs<T>, TDest> selector)
        {
            var mappedObservable = @this.RawObservable.Select(args =>
            {
                var result = selector(args);
                return new BasicConsumerArgs<TDest>(args.SyncChannel, result, args.OriginalArgs);
            });
            return new BasicConsumerObservable<TDest>(mappedObservable);
        }

        public BasicConsumerObservable<T> Where(
            Func<BasicConsumerArgs<T>, CancellationToken, ValueTask<bool>> predicate,
            bool multiple = false)
        {
            var filteredObservable = @this.RawObservable.Where(async (args, token) =>
            {
                var ret = await predicate(args, token);
                if (!ret)
                {
                    using var channelLock = await args.SyncChannel.GetChannelLockAsync(token);
                    await channelLock.Channel.BasicAckAsync(args.OriginalArgs.DeliveryTag, multiple, token);
                }
                return ret;
            });
            return @this with { RawObservable = filteredObservable };
        }

        public BasicConsumerObservable<T> Where(
            Func<BasicConsumerArgs<T>, bool> predicate,
            bool multiple = false)
        {
            var filteredObservable = @this.RawObservable.Where(async (args, token) =>
            {
                var ret = predicate(args);
                if (!ret)
                {
                    using var channelLock = await args.SyncChannel.GetChannelLockAsync(token);
                    await channelLock.Channel.BasicAckAsync(args.OriginalArgs.DeliveryTag, multiple, token);
                }
                return ret;
            });
            return @this with { RawObservable = filteredObservable };
        }

        public BasicConsumerObservable<T> Merge(BasicConsumerObservable<T> other)
        {
            var mergedObservable = @this.RawObservable.Merge(other.RawObservable);
            return @this with { RawObservable = mergedObservable };
        }
    }
}

This is just an idea, but you get the point. The heavy lifting is done by R3Async:

AsyncObservables handle subscriptions, async resource lifetimes, and backpressure.
You can integrate DI and scoped handlers by creating scopes inside your HandleAsync function.
Be creative — the core idea is that AsyncObservables map naturally to RabbitMQ streams, making it easy to build safe, composable, and fully asynchronous pipelines.
You can also convert them to IAsyncEnumerables and “return” them in SignalR hubs, Server-Sent Events (SSE), or other streaming endpoints.

AsyncR3 (Async reactive extensions for .NET)

Federico Alterio — Wed, 07 Jan 2026 16:10:48 +0000

GitHub: https://github.com/fedeAlterio/R3Async
NuGet: https://www.nuget.org/packages/R3Async
R3 (Sync Version) by Cysharp: https://github.com/Cysharp/R3

In modern .NET development, asynchronous programming has become essential for building responsive, scalable applications. While the traditional Reactive Extensions library (Rx.NET) has served developers well for years, it was designed in an era before async/await became ubiquitous. This creates friction with modern asynchronous code.

What is R3?

R3 is a modern reimplementation of Reactive Extensions for .NET, created by Cysharp. It has many improvements, but the most important one is to not stop the observable pipeline when an exception occurs.

R3Async: Bringing R3's Philosophy to Async observables

R3Async is the fully asynchronous counterpart to R3. While R3 focuses on synchronous reactive streams, R3Async reimagines the same philosophy with deeper async/await integration. It's a complete reimplementation built on top of ValueTask and IAsyncDisposable.

This implementation is better suited to common backend scenarios where a subscription wraps an external service (for example, a remote queue) rather than a local, in-process event source. In such cases, subscribing and unsubscribing must be genuinely asynchronous.

Background

The Async Gap

The fundamental problem with traditional Rx.NET is that Subscribe and Dispose are synchronous. In Rx.NET, when you call Subscribe() or dispose a subscription, these operations complete immediately and synchronously.

This creates a critical limitation: operators can't await the completion of cancellation. Consider the Switch operator, which should cancel the previous subscription when a new one arrives:

// Traditional Rx.NET - Switch can't await cancellation
observable
    .Switch()  // Initiates cancellation but doesn't wait for it!
    .Subscribe(x => Console.WriteLine(x));

// This can lead to overlapping operations because:
// 1. New observable starts immediately
// 2. Previous observable is only BEGINNING to cancel
// 3. Both may be running simultaneously for a brief period

Because IDisposable.Dispose() is synchronous, Rx.NET can only initiate cancellation—it can't wait for cleanup to complete. This can lead to race conditions, resource conflicts, and overlapping async operations.

Also, the fact that OnNext is synchronous doesn't allow the producer to handle backpressure properly, because consumers can't just block the thread under pressure.

Enter R3Async

R3Async solves this by making subscription, disposal and notifications inherently asynchronous. SubscribeAsync returns ValueTask<IAsyncDisposable>, allowing operators to truly await cleanup:

AsyncObservable

The heart of R3Async is the AsyncObservable<T> and AsyncObserver<T> abstract classes:

public abstract class AsyncObservable<T>
{
    public ValueTask<IAsyncDisposable> SubscribeAsync(
        AsyncObserver<T> observer, 
        CancellationToken cancellationToken);    
}

Key design decisions:

ValueTask Returns: Subscription is now async
IAsyncDisposable: Subscriptions return IAsyncDisposable for proper async cleanup
CancellationToken Support: First-class cancellation support throughout the API

AsyncObserver

Observers handle the stream events with three async methods:

public abstract class AsyncObserver<T> : IAsyncDisposable
{
    public ValueTask OnNextAsync(T value, CancellationToken cancellationToken);
    public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken);
    public ValueTask OnCompletedAsync(Result result);
}

Installation

dotnet add package R3Async

Basic Example

using R3Async;

// Create a simple observable that emits values every second
var subscription = await AsyncObservable
    .Interval(TimeSpan.FromSeconds(1))
    .Where(x => x % 2 == 0)
   .Select(x => x * 10)
    .SubscribeAsync(
        onNextAsync: async (value, ct) => 
        {
            Console.WriteLine($"Received: {value}");
        },
        cancellationToken: cancellationToken
    );

// Dispose when done
await subscription.DisposeAsync();

Common Operators

R3Async provides a rich set of operators for transforming and composing streams:

Transformation Operators

Select: Transform each value

var transformed = await AsyncObservable
    .Range(1, 5)
    .Select(async (x, ct) => 
    {
        await Task.Delay(10, ct);
        return x * 2;
    })
    .ToListAsync(cancellationToken);
// Result: [2, 4, 6, 8, 10]

Where: Filter values

var filtered = await AsyncObservable
    .Range(1, 10)
    .Where(x => x % 2 == 0)
    .ToListAsync(cancellationToken);
// Result: [2, 4, 6, 8, 10]

Scan: Accumulate values

var accumulated = await AsyncObservable
    .Range(1, 5)
    .Scan(0, (acc, x) => acc + x)
    .ToListAsync(cancellationToken);
// Result: [1, 3, 6, 10, 15]

Filtering Operators

Take: Take the first N items

var first3 = await AsyncObservable
    .Interval(TimeSpan.FromMilliseconds(100))
    .Take(3)
    .ToListAsync(cancellationToken);
// esult: [0, 1, 2]

Skip: Skip the first N items

var skipped = await AsyncObservable
    .Range(1, 10)
    .Skip(5)
    .ToListAsync(cancellationToken);
// Result: [6, 7, 8, 9, 10]

Distinct: Remove duplicates

var unique = await AsyncObservable
    .FromAsync([1, 2, 2, 3, 3, 3, 4])
    .Distinct()
    .ToListAsync(cancellationToken);
// Result: [1, 2, 3, 4]

Aggregation Operators

CountAsync: Count items

var count = await AsyncObservable
    .Range(1, 100)
    .Where(x => x % 2 == 0)
    .CountAsync(cancellationToken);
// esult: 50

FirstAsync / LastAsync: Get first or last item

var first = await AsyncObservable
    .Range(1, 10)
    .FirstAsync(cancellationToken);
// Result: 1

var last = await AsyncObservable
    .Range(1, 10)
    .LastAsync(cancellationToken);
// Result: 10

ToListAsync / ToDictionaryAsync: Materialize to collections

var list = await observable.ToListAsync(cancellationToken);
var dict = await observable.ToDictionaryAsync(
    x => x.Id, 
    x => x.Name, 
    cancellationToken);

Combination Operators

Merge: Merge multiple observables

var merged = AsyncObservable.Merge(
    AsyncObservable.Return(1),
    AsyncObservable.Return(2),
    AsyncObservable.Return(3)
);
var result = await merged.ToListAsync(cancellationToken);
// Result: [1, 2, 3] (order may vary)

Concat: Concatenate observables sequentially

var concatenated = AsyncObservable.Concat(
    AsyncObservable.Range(1, 3),
    AsyncObservable.Range(4, 3)
);
var result = await concatenated.ToListAsync(cancellationToken);
// Result: [1, 2, 3, 4, 5, 6]

Creating Custom Observables

R3Async provides several factory methods for creating observables:

FromAsync

// From Task
var taskObservable = AsyncObservable.FromAsync(async ct => 
{
    await Task.Delay(100, ct);
    return 42;
});

Factory Methods

// Single value
var single = AsyncObservable.Return(42);

// Empty stream
var empty = AsyncObservable.Empty<int>();

// ever completes
var never = AsyncObservable.Never<int>();

// Throw error
var error = AsyncObservable.Throw<int>(new Exception("Error"));

// Range of values
var range = AsyncObservable.Range(1, 100);

// Time-based
var interval = AsyncObservable.Interval(TimeSpan.FromSeconds(1));

Converting to IAsyncEnumerable and Backpressure

R3Async provides ToAsyncEnumerable to convert observables into IAsyncEnumerable<T>, enabling integration with async pull-based streams. However, this conversion introduces important backpressure considerations. R3Async requires you to explicitly provide a channel factory to define buffering semantics:

// Bounded channel - limits buffer size, applies backpressure
var boundedEnum = observable.ToAsyncEnumerable(
    () => Channel.CreateBounded<int>(new BoundedChannelOptions(10)
    {
        FullMode = BoundedChannelFullMode.Wait  // Blocks producer when full
    }));

// Unbounded channel - unlimited buffering (memory risk!)
var unboundedEnum = observable.ToAsyncEnumerable(
    () => Channel.CreateUnbounded<int>());

// Rendezvous channel - no buffering, direct handoff
var rendezvousEnum = observable.ToAsyncEnumerable(
    () => Channel.CreateBounded<int>(0));

This explicit design forces you to think about backpressure strategies:

Bounded channels apply backpressure by blocking the producer when the buffer is full, preventing memory exhaustion but potentially slowing down fast sources
Unbounded channels never block but can consume unlimited memory if the consumer is slower than the producer
DropOldest/DropNewest strategies discard items when full, useful for scenarios where only the latest data matters

Example with backpressure handling:

// Fast producer, slow consumer scenario
var fastProducer = AsyncObservable.Interval(TimeSpan.FromMilliseconds(10));

// Convert with bounded buffer and backpressure
var asyncEnum = fastProducer.ToAsyncEnumerable(
    () => Channel.CreateBounded<long>(new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.Wait  // Apply backpressure
    }));

// Slow consumer - the producer will automatically slow down
await foreach (var item in asyncEnum.WithCancellation(cancellationToken))
{
    await Task.Delay(100, cancellationToken);  // Slow processing
    Console.WriteLine(item);
}
// Producer is throttled by backpressure - no memory overflow!

This design makes esource management decisions explicit rather than hiding them behind default behaviors.

Angular Signals, But Better Now for .NET

Federico Alterio — Mon, 15 Dec 2025 20:15:45 +0000

C# is Great for async

Did you know that C# is where async-await and reactive extensions was born?

Before rxjs made Observable popular in JavaScript, C# already had IObservable — complete with powerful LINQ operators (without that pipe cerimony, thanks to extension methods).
The same goes for promises and abort signals: they’re essentially JavaScript re-implementations of Task and CancellationToken.

C# has some of the best async support out there. In fact, it still offers capabilities that JavaScript lacks — async locals, custom awaitables.

And yet, one thing has been missing..

Signals.

That’s where SignalsDotnet comes in.

SignalsDotnet

SignalsDotnet brings the mental model and ergonomics of Angular Signals into the .NET ecosystem — without abandoning what makes C# great. It’s a modern reactive primitive designed for .NET developers who are tired of boilerplate-heavy INotifyPropertyChanged, leaky subscriptions, and over-engineered state management.

At its core, a signal is just a value — but one that knows who depends on it.

When a signal changes, everything that derives from it updates automatically. No manual wiring. No += and -=.
C#

var a = new Signal<int>(2);
var b = new Signal<int>(3);

var sum = Signal.Computed(() => a.Value + b.Value);

Console.WriteLine(sum.Value); // 5

a.Value = 10;

Console.WriteLine(sum.Value); // 13

Here’s the (quasi) equivalent minimal JavaScript example using Angular-style signals:

Angular

const a = signal(2);
const b = signal(3);

const sum = computed(() => a() + b());

Angular signals are tied to Angular’s lifecycle and scheduling.
SignalsDotnet is framework-agnostic: everything is computed synchronously and immediately, with no hidden batching or render cycle — closer to an Rx Subject.

Async computed

Do you remember when I said C# has one of the best async support?
What happens if our computation function is async?
Look at this:

var a = new Signal<int>(2);
var b = new Signal<int>(3);

var computed = Signal.AsyncComputed(async ct =>
{
    // Access signal BEFORE await
    int first = a.Value;

    // Simulate async work
    await Task.Delay(500, ct);

    // Access signal AFTER await
    int second = b.Value;

    return first + second;
},  defaultValue: 0,
    ConcurrentChangeStrategy.CancelCurrent);

You can notice that the b signal is accessed after the await. This is possible because C# has the concept of ExecutionContext and AsyncLocal, which effectively gives a “context” to our async call stack.

What this means in practice: even if the thread is doing other work while awaiting, C# still knows the computation depends on both a and b. The dependency tracking survives the suspension of the async method.

In contrast, JavaScript signals or Angular resources lose that kind of implicit context across async boundaries
This is an attempt to replicate an async computed in Angular, using resources
Angular

const a = signal(2);
const b = signal(3);

const computed = resource({
  params: () => ({ a: a() }), // only tracks a before async

  loader: async ({ params, abortSignal }) => {
    // Access signal BEFORE await
    const first = params.a;

    // Simulate async work
    await new Promise((resolve) => setTimeout(resolve, 500));

    // Access signal AFTER await
    const second = b(); // this access is NOT tracked automatically

    return first + second;
  },
  initialValue: 0,
});

To make this work in Angular, you would have to either:

Declare the dependency on b before it’s accessed in the async function,
Split the computation into multiple computed signals or resources, or
Fallback to RxJS for more flexible dynamic tracking.

All of these approaches defeat a bit the purpose of computed signals, which are supposed to automatically track dependencies without extra boilerplate.

C# would have no problem to track also in while loops, for loops or whatever, it just works.

Can you imagine doing that with angular resources?

var numbers = new[]
{
    new Signal<int>(1),
    new Signal<int>(2),
    new Signal<int>(3)
};

// AsyncComputed that sums all signals, with async/await inside the loop
var asyncSum = Signal.AsyncComputed(async ct =>
{
    int total = 0;
    for (int i = 0; i < numbers.Length; i++)
    {
        total += numbers[i].Value;       // dependency tracked
        await Task.Delay(100, ct);       // async work inside loop
    }
    return total;
}, defaultValue: 0, ConcurrentChangeStrategy.CancelCurrent);

There’s no easy counterpart in Angular — even in RxJS, because loops become recursion.

SignalsDotnet: Signal Types at a Glance

Signal Type	Description	Example
`Signal<T>`	Writable, reactive value	`var count = new Signal<int>(0);`
`IReadOnlySignal<T>`	Read-only / computed signal	`var fullName = Signal.Computed(() => $"{first.Value} {last.Value}");`
`IAsyncReadOnlySignal<T>`	Async computed signal	`var asyncSum = Signal.AsyncComputed(async ct => { await Task.Delay(100); return count.Value * 2; });`
`ISignal<T>` (Linked)	Writable computed signal that can be overridden	`var linked = Signal.Linked(() => source.Value * 2); linked.Value = 100;`
`CollectionSignal<TObservableCollection>`	Deeply reactive wrapper around an `ObservableCollection<T>`	`var people = new CollectionSignal<ObservableCollection<Person>>();`
`DictionarySignal<TKey, TValue>`	Reactive dictionary with fine‑grained key tracking	`var scores = new DictionarySignal<string, int>(); scores["a"] = 1;`
`Effect`	Runs side effects when dependencies change	`Signal.Effect(() => Console.WriteLine(count.Value));`

Highlights:

Automatic dependency tracking — signals automatically subscribe to only the signals they use.
Supports computed, async computed with cancellation, and linked writable computed signals.
CollectionSignal listens not just to the collection reference, but also to its internal changes (add/remove/clear).
DictionarySignal tracks keys independently so computed signals only rerun when the specific keys they depend on change.
Fully framework‑agnostic — works in console apps, WPF, MAUI, Blazor, Unity, Godot, etc.

Bonus: All SignalsDotnet signals are Observables. That means they integrate seamlessly with all Rx (R3), giving you the full power of operators like Select, Where, CombineLatest, and more.

Any feedback, suggestions, or experiences are more than welcome — I hope you give SignalsDotnet a try!