如何重新訂閱特定點的序列？

Question

我試圖解決以下問題：a）用戶從IObservable接收事件一段時間。然後它取消訂閱，做一些東西，然後再次訂閱。在這裏它應該開始接收來自完全取消訂閱的同一點的事件。 b）這種行爲對於多用戶模型是可取的。例如。當一個人取消訂閱時，其他人應該繼續接收事件。

有沒有RX方面的建議？

在此先感謝！

Answer 1

這聽起來像你需要一個「pausable」流。假設一次只有一個用戶會處理這些值（而其他用戶只是等待），這個solution可能就是你需要的。

Answer 2

這裏有一個相當簡單的Rx方式來做你想從my answer複製到this other question。我創建了一個名爲Pausable的擴展方法，該方法需要可觀察的源代碼和第二個可觀察的布爾值，以暫停或恢復可觀察值。

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); 
    }); 
} 

它可以像這樣使用：

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);