.Net Reactive extensions: Remove Replay

1
votes

In the below code, I am using Replay() to avoid price update loss while I am doing stuff [...]. I need to send these price update after stuff[...] is completed.

var observable = Observable.FromEventPattern<Price>(h => book.PriceUpdated += h, h => book.PriceUpdated -= h)
                           .Replay();
observable.Connect();

// Do some stuff [...]

observable.Select(p => Observable.FromAsync(() => SendPriceUpdateAsync(p.Sender, p.EventArgs, socket)))
                      .Concat()
                      .Subscribe();

I want to remove the Replay() after the stuff [...] is done and the price sent (I don't want all the values to be stored and I don't need the values anymore once I have sent them).

Is there a simple way to do it ?

2
c#reactive-programmingsystem.reactive
I'm not sure I understand your question. What exactly are you trying to achieve? Do you wish to stop listening to the events once you've sent all the prices? Do you want to keep handling the price updates after you've sent the prices once? - Alex Pshul
Before the 'stuff[...]' I keep track of the price update. After the 'stuff[...]' I need to send all the price update in order, without loosing any (even those before 'stuff[...]'). Once I did this I only need to send the price update to come, I don't need the Replay() anymore (which will store all event to be able to replay them) - Lou

2 Answers

3
votes

So what you actually want is to delay all events until you are finished with Do some stuff. I suggest to not use Replay at all.

Instead try an approach like this:

var observable = eventObservable
.TakeUntil(Observable.Start(() => { /* Do some stuff [...] */ })) //collect only events while processing, completes the observable when done processing
.ToList() //with ToList only propagate all at once on completion
.SelectMany(list => list.ToObservable()) //unfold the list to single events again
.Concat(eventObservable); //continue with normal events observable
1
votes

So I couldn't find any way to do it with operators, but we can always create our own connectable observable class! :)

public class UntilSubscribeReplay<T> : IObservable<T>
{
    private readonly IObservable<T> _parent;

    private HashSet<IObserver<T>> _observers = new HashSe
    private IDisposable _subscription;
    private readonly Queue<T> _itemsQueue = new Queue<T>(

    public UntilSubscribeReplay(IObservable<T> parent)
    {
        _parent = parent;
        Subscribe();
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        if (_subscription == null)
            Subscribe();

        _observers.Add(observer);
        EmitQueuedItems();

        return Disposable.Create(() =>
        {
            _observers.Remove(observer);
            if (_observers.Count == 0)
                _subscription.Dispose();
        });
    }

    private void Subscribe()
    {
        _subscription = _parent
            .Do(_itemsQueue.Enqueue)
            .Subscribe(_ => EmitQueuedItems());
    }

    private void EmitQueuedItems()
    {
        if (_observers.Count == 0)
            return;

        while (_itemsQueue.TryDequeue(out T item))
        {
            foreach (IObserver<T> observer in _observers)
            {
                observer.OnNext(item);
            }
        }
    }
}

And of course, an extension method to use for your convenience. :)

public static class MyObservableExtensions
{
    public static IObservable<T> UntilSubscribeReplay<T>(this IObservable<T> source) 
        => new UntilSubscribeReplay<T>(source);
}

And here is how you would use it:

IObservable<int> myReplay = myObservable.UntilSubscribeReplay();

// Do Stuff

myReplay.Subscribe(value => HandleValue(value));

Note, that there is no need to use the connect. The proper way would be to implement it as a Connectable Observable (To avoid memory leaks when you use the extension but don't subscribe eventually), but for your specific case, it should be enough.

Let me know how it goes. :)