I need an Rx.NET observable that accumulates items while there is no active subscribers, and emits the whole accumulated sequence (and any future items) to new subscribes, as soon as there is any.
It is different from ReplaySubject
in that it doesn't keep the items which have been once replayed to any subscriber. Thus, once a queued item has been observed by the current subscribers, it gets removed from the queue and won't be seen by any new future subscribers.
Can something like that be composed using standard Rx.NET operators?
I need it to tackle races conditions in the following scenario. There is a looping async workflow RunWorkflowAsync
which needs to perform ResetAsync
task when it observes a specific ResetRequestedEvent
message.
Here's the whole thing as a .NET 6 console app:
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading.Channels;
try
{
await TestAsync();
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
async Task TestAsync()
{
var resetRequestsSubject = new Subject<ResetRequestedEvent>();
using var cts = new CancellationTokenSource(20000);
await Task.WhenAll(
SimulateResetRequests(cts.Token),
RunWorkflowAsync(resetRequestsSubject, cts.Token));
// simulate emmiting reset requests
async Task SimulateResetRequests(CancellationToken cancelToken)
{
async Task Raise(int n, int delay)
{
var ev = new ResetRequestedEvent(n);
Console.WriteLine($"{ev} issued");
resetRequestsSubject!.OnNext(ev);
await Task.Delay(delay, cancelToken);
}
await Raise(1, 50);
await Raise(2, 50);
await Raise(3, 50);
await Raise(4, 1000);
await Raise(5, 5000);
await Raise(6, 4000);
await Raise(7, 3000);
resetRequestsSubject.OnCompleted();
}
// simulate the reset task
async Task ResetAsync(CancellationToken cancelToken)
{
await Task.Delay(1000, cancelToken);
Console.WriteLine("Reset done");
}
// simulate the work task
async Task DoWorkAsync(CancellationToken cancelToken)
{
await Task.Delay(2000, cancelToken);
Console.WriteLine("Work done");
}
// do reset, then work in a loop until cancelled
async Task RunWorkflowAsync(IObservable<ResetRequestedEvent> resetRequests, CancellationToken externalCancelToken)
{
// from this point, make sure reset requests don't go unobserved
var accumulatedResetRequests = resetRequests.Accumulate(externalCancelToken);
using var auto1 = accumulatedResetRequests.Connect();
while (true)
{
externalCancelToken.ThrowIfCancellationRequested(); // stops the whole workflow
using var internalCts = CancellationTokenSource.CreateLinkedTokenSource(externalCancelToken);
var internalCancelToken = internalCts.Token;
// signal cancellation upon the most recent reset request
using var auto2 = accumulatedResetRequests
.Do(ev => Console.WriteLine($"{ev} seen"))
.Throttle(TimeSpan.FromMilliseconds(100))
.Do(ev => Console.WriteLine($"{ev} acted upon"))
.Subscribe(_ => internalCts.Cancel());
try
{
// start with a reset
await ResetAsync(internalCancelToken);
// do work until another reset is requested
while (true)
{
await DoWorkAsync(internalCancelToken);
}
}
catch (OperationCanceledException)
{
}
}
}
}
record ResetRequestedEvent(int Number);
public static class RxExt
{
class CumulativeObservable<T> : IConnectableObservable<T>
{
readonly IObservable<T> _source;
readonly Channel<T> _channel;
readonly CancellationToken _cancelToken;
public CumulativeObservable(IObservable<T> source, CancellationToken cancellationToken)
{
_source = source;
_channel = Channel.CreateUnbounded<T>();
_cancelToken = cancellationToken;
}
public IDisposable Connect() =>
_source.Subscribe(
onNext: item => _channel.Writer.TryWrite(item),
onError: ex => _channel.Writer.Complete(ex),
onCompleted: () => _channel.Writer.Complete());
public IDisposable Subscribe(IObserver<T> observer) =>
_channel.Reader.ReadAllAsync(_cancelToken).ToObservable().Subscribe(observer);
}
public static IConnectableObservable<T> Accumulate<T>(
this IObservable<T> @this,
CancellationToken cancellationToken) =>
new CumulativeObservable<T>(@this, cancellationToken);
}
The idea is to stop all pending tasks inside RunWorkflowAsync
and perform ResetAsync
when ResetRequestedEvent
message comes along.
I realize there's more than one way to cook an egg (and implement RunWorkflowAsync
), but I like this approach as I don't need to think about thread safety when I use and recycle internalCts
cancellation token source (to stop all pending task before another iteration).
Above, CumulativeObservable
does what I want, but it's a very naive implementation which only supports one concurrent observable (unlike, say, ReplaySubject
) and lacks any safety checks.
I'd prefer a composition that can be built using standard operators.
ReplayOnce
operator that I have posted here. Have we discussed about that in the past? Somehow this topic sounds familiar to me!