Pause and Resume Subscription on cold IObservable

Question

Using Rx, I desire pause and resume functionality in the following code:

How to implement Pause() and Resume() ?

    static IDisposable _subscription;

    static void Main(string[] args)
    {
        Subscribe();
        Thread.Sleep(500);
        // Second value should not be shown after two seconds:
        Pause();
        Thread.Sleep(5000);
        // Continue and show second value and beyond now:
        Resume();
    }

    static void Subscribe()
    {
        var list = new List<int> { 1, 2, 3, 4, 5 };
        var obs = list.ToObservable();
        _subscription = obs.SubscribeOn(Scheduler.NewThread).Subscribe(p =>
        {
            Console.WriteLine(p.ToString());
            Thread.Sleep(2000);
        },
        err => Console.WriteLine("Error"),
        () => Console.WriteLine("Sequence Completed")
        );
    }

    static void Pause()
    {
        // Pseudocode:
        //_subscription.Pause();
    }

    static void Resume()
    {
        // Pseudocode:
        //_subscription.Resume();
    }

Rx Solution?

• I believe I could make it work with some kind of Boolean field gating combined with thread locking (Monitor.Wait and Monitor.Pulse)

• But is there an Rx operator or some other reactive shorthand to achieve the same aim?

Answer 1

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.

Answer 2

Here is my answer. I believe there may be a race condition around pause resume, however this can be mitigated by serializing all activity onto a scheduler. (favor Serializing over synchronizing).

using System;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using Microsoft.Reactive.Testing;
using NUnit.Framework;

namespace StackOverflow.Tests.Q7620182_PauseResume
{
    [TestFixture]
    public class PauseAndResumeTests
    {
        [Test]
        public void Should_pause_and_resume()
        {
            //Arrange
            var scheduler = new TestScheduler();

            var isRunningTrigger = new BehaviorSubject<bool>(true);
            Action pause = () => isRunningTrigger.OnNext(false);
            Action resume = () => isRunningTrigger.OnNext(true);

            var source = scheduler.CreateHotObservable(
                ReactiveTest.OnNext(0.1.Seconds(), 1),
                ReactiveTest.OnNext(2.0.Seconds(), 2),
                ReactiveTest.OnNext(4.0.Seconds(), 3),
                ReactiveTest.OnNext(6.0.Seconds(), 4),
                ReactiveTest.OnNext(8.0.Seconds(), 5));

            scheduler.Schedule(TimeSpan.FromSeconds(0.5), () => { pause(); });
            scheduler.Schedule(TimeSpan.FromSeconds(5.0), () => { resume(); });



            //Act
            var sut = Observable.Create<IObservable<int>>(o =>
            {
                var current = source.Replay();
                var connection = new SerialDisposable();
                connection.Disposable = current.Connect();

                return isRunningTrigger
                    .DistinctUntilChanged()
                    .Select(isRunning =>
                    {
                        if (isRunning)
                        {
                                //Return the current replayed values.
                                return current;
                        }
                        else
                        {
                                //Disconnect and replace current.
                                current = source.Replay();
                                connection.Disposable = current.Connect();
                                //yield silence until the next time we resume.
                                return Observable.Never<int>();
                        }

                    })
                    .Subscribe(o);
            }).Switch();

            var observer = scheduler.CreateObserver<int>();
            using (sut.Subscribe(observer))
            {
                scheduler.Start();
            }

            //Assert
            var expected = new[]
            {
                    ReactiveTest.OnNext(0.1.Seconds(), 1),
                    ReactiveTest.OnNext(5.0.Seconds(), 2),
                    ReactiveTest.OnNext(5.0.Seconds(), 3),
                    ReactiveTest.OnNext(6.0.Seconds(), 4),
                    ReactiveTest.OnNext(8.0.Seconds(), 5)
                };
            CollectionAssert.AreEqual(expected, observer.Messages);
        }
    }
}

Answer 3

Here it is as an application of IConnectableObservable that I corrected slightly for the newer api (original here):

public static class ObservableHelper {
    public static IConnectableObservable<TSource> WhileResumable<TSource>(Func<bool> condition, IObservable<TSource> source) {
        var buffer = new Queue<TSource>();
        var subscriptionsCount = 0;
        var isRunning = System.Reactive.Disposables.Disposable.Create(() => {
            lock (buffer)
            {
                subscriptionsCount--;
            }
        });
        var raw = Observable.Create<TSource>(subscriber => {
            lock (buffer)
            {
                subscriptionsCount++;
                if (subscriptionsCount == 1)
                {
                    while (buffer.Count > 0) {
                        subscriber.OnNext(buffer.Dequeue());
                    }
                    Observable.While(() => subscriptionsCount > 0 && condition(), source)
                        .Subscribe(
                            v => { if (subscriptionsCount == 0) buffer.Enqueue(v); else subscriber.OnNext(v); },
                            e => subscriber.OnError(e),
                            () => { if (subscriptionsCount > 0) subscriber.OnCompleted(); }
                        );
                }
            }
            return isRunning;
        });
        return raw.Publish();
    }
}

Answer 4

It just works:

    class SimpleWaitPulse
    {
      static readonly object _locker = new object();
      static bool _go;

      static void Main()
      {                                // The new thread will block
        new Thread (Work).Start();     // because _go==false.

        Console.ReadLine();            // Wait for user to hit Enter

        lock (_locker)                 // Let's now wake up the thread by
        {                              // setting _go=true and pulsing.
          _go = true;
          Monitor.Pulse (_locker);
        }
      }

      static void Work()
      {
        lock (_locker)
          while (!_go)
            Monitor.Wait (_locker);    // Lock is released while we’re waiting

        Console.WriteLine ("Woken!!!");
      }
    }

Please, see How to Use Wait and Pulse for more details