Pause and Resume Subscription on cold IObservable

by

Here’s a reasonably simple Rx way to do what you want. I’ve created an extension method called Pausable that takes a source observable and a second observable of boolean that pauses or resumes the observable.

public static IObservable<T> Pausable<T>(
    this IObservable<T> source,
    IObservable<bool> pauser)
{
    return Observable.Create<T>(o =>
    {
        var paused = new SerialDisposable();
        var subscription = Observable.Publish(source, ps =>
        {
            var values = new ReplaySubject<T>();
            Func<bool, IObservable<T>> switcher = b =>
            {
                if (b)
                {
                    values.Dispose();
                    values = new ReplaySubject<T>();
                    paused.Disposable = ps.Subscribe(values);
                    return Observable.Empty<T>();
                }
                else
                {
                    return values.Concat(ps);
                }
            };

            return pauser.StartWith(false).DistinctUntilChanged()
                .Select(p => switcher(p))
                .Switch();
        }).Subscribe(o);
        return new CompositeDisposable(subscription, paused);
    });
}

It can be used like this:

var xs = Observable.Generate(
    0,
    x => x < 100,
    x => x + 1,
    x => x,
    x => TimeSpan.FromSeconds(0.1));

var bs = new Subject<bool>();

var pxs = xs.Pausable(bs);

pxs.Subscribe(x => { /* Do stuff */ });

Thread.Sleep(500);
bs.OnNext(true);
Thread.Sleep(5000);
bs.OnNext(false);
Thread.Sleep(500);
bs.OnNext(true);
Thread.Sleep(5000);
bs.OnNext(false);

It should be fairly easy for you to put this in your code with the Pause & Resume methods.

More Related Contents:

  1. If async-await doesn’t create any additional threads, then how does it make applications responsive?
  2. What is the difference between asynchronous programming and multithreading?
  3. Implement C# Generic Timeout
  4. How to protect resources that may be used in a multi-threaded or async environment?
  5. Async process start and wait for it to finish
  6. “async Task then await Task” vs “Task then return task” [duplicate]
  7. Good introduction to the .NET Reactive Framework [closed]
  8. Benefits of using async and await keywords
  9. Named Mutex with await
  10. What thread runs the code after the `await` keyword?
  11. Calling async method on button click
  12. Calling async methods from non-async code
  13. Is .GetAwaiter().GetResult(); safe for general use?
  14. C# Threading/Async: Running a task in the background while UI is interactable
  15. Async thread body loop, It just works, but how?
  16. Async WebApi Thread.CurrentCulture
  17. I thought await continued on the same thread as the caller, but it seems not to
  18. How do I create a custom SynchronizationContext so that all continuations can be processed by my own single-threaded event loop?
  19. Synchronous implementation of interface that returns Task
  20. What is the difference between .Wait() vs .GetAwaiter().GetResult()?
  21. Write to a file from multiple threads asynchronously c#
  22. How to call back async function from Rx Subscribe?
  23. Async and Await with For Loop [duplicate]
  24. Calling an async method from a synchronous method
  25. Calling TaskCompletionSource.SetResult in a non blocking manner
  26. How to pass parameters to ThreadStart method in Thread?
  27. Code for a simple thread pool in C# [closed]
  28. Process files concurrently and asynchronously
  29. How to keep a .NET console app running?
  30. Why is this web api controller not concurrent?

Leave a Comment