0

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.

13
  • 1
    stackoverflow.com/a/24805489/161739 for ideas perhaps, however I would never go as far as implementing IObservable, standard operators wit synchronized Subjects which obey the RX contracts can model every case out there Commented May 23, 2022 at 3:49
  • 1
    values that never been observed emmited from this operator stackoverflow.com/questions/60483181/… Commented May 23, 2022 at 3:52
  • 1
    "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." -- If I understand correctly the queue will only be used when there are zero subscribers. As soon as the first subscriber subscribes, it will consume the whole queue immediately, and after that moment the queue could be discarded. It should be instantiated once again when the last subscriber unsubscribes, and the subscribers become zero again. Is my understanding correct? Commented May 23, 2022 at 12:10
  • 1
    @TheodorZoulias, correct! The (empty) queue can be discarded until there is zero subscribers again. Then the queue becomes again relevant. In the code snippet I got it working like that for single subscriber (using a channel). I imagine it'd be a lot more involved for more general scenario where multiple subscribers can connect and disconnect dynamically :)
    – noseratio
    Commented May 23, 2022 at 12:53
  • 1
    I see. What you want is probably quite close to a ReplayOnce operator that I have posted here. Have we discussed about that in the past? Somehow this topic sounds familiar to me! Commented May 23, 2022 at 13:40

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.