RoomSharp.Reactive 0.5.5

RoomSharp.Reactive

Lightweight reactive operators, computed views, query caching, UI collection binding, and Rx.NET adapters for RoomSharp.

Installation

dotnet add package RoomSharp.Reactive

Observe + debounce

using RoomSharp.Reactive;

var todos = db.GetTableIdOrThrow("todos");

// Default debounce uses ReactiveDefaults.DefaultDebounceInterval (150ms)
var query = db.ObserveReactive(
    ct => new ValueTask<IReadOnlyList<Todo>>(db.TodoDao.GetAllAsync()),
    new ReactiveQueryOptions
    {
        DebounceInterval = TimeSpan.FromMilliseconds(200)
    },
    todos);

using var subscription = query.Subscribe(
    list => Render(list),
    error => Log(error));

QueryCache (single-flight + TTL)

using RoomSharp.Reactive;

var cache = new QueryCache(db, new QueryCacheOptions
{
    DefaultTtl = TimeSpan.FromSeconds(30),
    MaxEntries = 1_000,
    FactoryCancellationMode = CacheFactoryCancellationMode.NeverCancelInFlight
});
var key = db.BuildCacheKey<IReadOnlyList<Todo>>("select * from todos", parameters: null);

var cached = db.ObserveCached(
    cache,
    key,
    ct => new ValueTask<IReadOnlyList<Todo>>(db.TodoDao.GetAllAsync()),
    new QueryCacheEntryOptions
    {
        Ttl = TimeSpan.FromSeconds(10),
        TableIds = new[] { db.GetTableIdOrThrow("todos") }
    },
    options: null,
    db.GetTableIdOrThrow("todos"));

using var sub = cached.Subscribe(
    list => Render(list),
    error => Log(error));

QueryCache uses single-flight execution for concurrent requests with the same key, TTL expiration, table invalidation, and a bounded entry count. Completed entries over MaxEntries are evicted from the oldest-accessed entries first; in-flight factories are not evicted.

FactoryCancellationMode controls how caller cancellation affects the shared factory:

• NeverCancelInFlight keeps the shared factory running even if one waiter cancels. This is the default and is usually best for UI screens.
• LinkedToFirstCaller links the factory to the first caller cancellation token. Use it only when cancelled first callers should abandon shared work.

For operational telemetry, implement IQueryCacheMetrics2 to receive eviction reasons, active entry counts, and factory duration/failure callbacks.

ComputedView (derived reactive value)

using RoomSharp.Reactive;

var todosQuery = db.ObserveReactive(
    ct => new ValueTask<IReadOnlyList<Todo>>(db.TodoDao.GetAllAsync()),
    db.GetTableIdOrThrow("todos"));

var stats = ComputedView.Combine(
    todosQuery,
    todos => new TodoStats(todos.Count),
    new ComputedViewOptions
    {
        DebounceInterval = TimeSpan.FromMilliseconds(200)
    });

using var sub = stats.Subscribe(
    value => UpdateStats(value),
    error => Log(error));

ObservableCollection binding (WPF/WinUI)

using RoomSharp.Reactive;
using System.Collections.ObjectModel;
using System.Threading;

var collection = new ReactiveObservableCollection<Todo>();
var ui = SynchronizationContext.Current!;

using var sub = query
    .AsObservable()
    .BindToObservableCollection(
        collection,
        new ReactiveCollectionBindingOptions<Todo>
        {
            Dispatch = action => ui.Post(_ => action(), null),
            KeySelector = todo => todo.Id,
            UpdateMode = ReactiveCollectionUpdateMode.Reset,
            MergeExisting = (existing, incoming) =>
            {
                existing.Title = incoming.Title;
                existing.IsDone = incoming.IsDone;
            },
            OnError = error => Log(error)
        });

ReactiveObservableCollection<T> can replace or merge a full result set with one reset notification. This is useful for larger UI lists where per-item collection notifications create unnecessary UI work.

Diagnostics

Reactive refreshes, cache factories, and collection binding updates publish ActivitySource spans through RoomSharp.Reactive:

using System.Diagnostics;

using var listener = new ActivityListener
{
    ShouldListenTo = source => source.Name == ReactiveDiagnostics.ActivitySourceName,
    Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllDataAndRecorded
};

ActivitySource.AddActivityListener(listener);

The activities include tags such as table count, subscriber count, item count, cache key hash, active entries, and factory duration.

Lightweight operators

using RoomSharp.Reactive;

var stream = query
    .AsObservable()
    .DistinctUntilChanged()
    .Throttle(TimeSpan.FromMilliseconds(150))
    .Buffer(TimeSpan.FromMilliseconds(250));

Rx.NET integration

using RoomSharp.Reactive.Rx;

var observable = query.ToObservable();
observable
    .Throttle(TimeSpan.FromMilliseconds(300))
    .DistinctUntilChanged()
    .Subscribe(list => Render(list));

Requirements

• RoomSharp 0.4.3+
• System.Reactive 6.0+ is referenced transitively by this package for Rx.NET adapters.