Observables

Virtufin.Data provides IObservable<T> adapters for three reactive data sources: NATS pub/sub, WebSocket frames, and Valkey keyspaces. All three implement auto-reconnect with exponential backoff.

The `IObservable<T>` contract

public interface IObservable<out T>
{
    IDisposable Subscribe(IObserver<T> observer);
}

public interface IObserver<in T>
{
    void OnNext(T value);
    void OnError(Exception ex);
    void OnCompleted();
}

The adapters in Virtufin.Data wrap an external stream and present it as an IObservable<T>. Subscribers use Subscribe(IObserver<T>) to receive values; unsubscribing disposes the underlying subscription.

NATS observable

using Virtufin.Data;

var nats = new NatsConnection("nats://localhost:4222");
var observable = new NatsObservable<string>(nats, "orders.>");

using var sub = observable.Subscribe(new ConsoleObserver<string>());

class ConsoleObserver<T> : IObserver<T>
{
    public void OnNext(T value) => Console.WriteLine(value);
    public void OnError(Exception ex) => Console.Error.WriteLine(ex);
    public void OnCompleted() => Console.WriteLine("done");
}

The NatsObservable: - Subscribes to subjects matching the pattern - Re-subscribes on reconnect (the pattern, not the exact subject) - Buffers the last N values for late subscribers (configurable)

Note: NatsObservable connects to NATS directly — a library-level data source, not a managed Dapr component. Workers running inside the WorkManager MUST use the API gateway's Pubsub.Subscribe RPC for pub/sub rather than connecting to NATS directly. Direct NATS connections are intended for out-of-process consumers (library-level code, back-testing, data replay).

WebSocket observable

var ws = new ClientWebSocket();
await ws.ConnectAsync(new Uri("wss://api.binance.com/ws/btcusdt@trade"), CancellationToken.None);
var observable = new WebSocketObservable<TradeEvent>(ws);

using var sub = observable.Subscribe(observer);

The WebSocketObservable: - Reads frames from the WebSocket - Parses each frame (configurable: JSON, FlatBuffers, raw) - Reconnects on disconnect with exponential backoff - Surfaces parse errors as OnError

Valkey observable

var mux = await ConnectionMultiplexer.ConnectAsync("localhost:6379");
var db = mux.GetDatabase();
var observable = new ValkeyObservable<DecimalAmount>(db, "btc-price");

using var sub = observable.Subscribe(observer);

The ValkeyObservable: - Uses Valkey keyspace notifications (__keyspace@0__:) to watch for changes - Re-subscribes on PSUB reconnect - Decodes values (configurable: FlatBuffers, JSON, raw bytes)

Prerequisite: Keyspace notifications are disabled by default on both Valkey and Redis. Enable them on the server with: CONFIG SET notify-keyspace-events KA or add notify-keyspace-events KA to your valkey.conf/redis.conf. Without this, ValkeyObservable will never fire — the server never emits the keyspace events it needs.

Why `IObservable<T>`?

The .NET reactive extensions (System.Reactive) are the idiomatic way to express asynchronous streams. Every consumer of streaming data in the platform already uses IObservable<T> — the adapters integrate without glue code.

For more complex stream combinators (filter, merge, throttle), use System.Reactive:

var fastTrades = observable
    .Where(t => t.Symbol == "BTCUSDT")
    .Throttle(TimeSpan.FromMilliseconds(100))
    .Select(t => t.Price);

Performance

NATS: native protocol, single connection per subject pattern, no serialization overhead for string/byte[] payloads.
WebSocket: one connection per WebSocketObservable, frame-by-frame parsing.
Valkey: uses connection multiplexing (one ConnectionMultiplexer can serve many ValkeyObservable instances).

For very high-throughput paths, use the byte[] overload of each observable and parse downstream — that avoids intermediate allocations.