從合併序列中刪除IObservable

Question

我正在使用Observable.Merge合併多個序列並在UI中的單個視圖中顯示。用戶可以選擇添加或刪除UI中的序列（供稿）。雖然我已經使用合併來合併提要。我不知道如何從合併序列中分離IObservable。目前，我正在創建一個全新的IObservable，省略我想要的提要。是否可以動態添加和刪除ViewModel已經訂閱的IObsevable？

Answer 1

這可能會幫助你開始，但在重新連接，由於B是冷的，B將重新從頭開始：

var a = Observable.Generate('A', x => x <= Char.MaxValue, x => ++x, x => x, x => TimeSpan.FromMilliseconds(200)).Select(x => "a: " + x).Publish(); 
var b = Observable.Generate('a', x => x <= Char.MaxValue, x => ++x, x => x, x => TimeSpan.FromMilliseconds(500)).Select(x => "b: " + x).Publish(); 

var merged = a.Merge(b).Publish(); 
var submerged = merged.Subscribe(x => x.Dump()); 

var subA = a.Connect(); 
var subB = b.Connect(); 
merged.Connect(); 

Task.Delay(2000).ContinueWith(t => subB.Dump("Disposing b.").Dispose()); 
Task.Delay(4000).ContinueWith(t => b.Connect()).ContinueWith(_ => "Reconnected to b"); 

編輯：

要添加另一個 'C' 的合併IO：

var a = Observable.Generate('A', x => x <= Char.MaxValue, x => ++x, x => x, x => TimeSpan.FromMilliseconds(200)).Select(x => "a: " + x).Publish(); 
var b = Observable.Generate('a', x => x <= Char.MaxValue, x => ++x, x => x, x => TimeSpan.FromMilliseconds(500)).Select(x => "b: " + x).Publish(); 
var c = Observable.Generate('1', x => x <= Char.MaxValue, x => ++x, x => x, x => TimeSpan.FromMilliseconds(100)).Select(x => "c: " + x).Publish(); 

var merged = a.Merge(b).Merge(c).Publish(); 
var submerged = merged.Subscribe(x => x.Dump()); 

var subA = a.Connect(); 
var subB = b.Connect(); 
merged.Connect(); 

Task.Delay(2000).ContinueWith(t => subB.Dump("Disposing b.").Dispose()); 
Task.Delay(4000).ContinueWith(t => b.Connect()).ContinueWith(_ => "Reconnected to b".Dump()); 
Task.Delay(6000).ContinueWith(t => c.Connect()).ContinueWith(_ => "Connecting to c".Dump()); 

Answer 2

在使用IObservable<IObservable<T>>看一看，然後使用Merge。這會自動讓您通過結束內部IObservable<T>來刪除序列。簡單。

Answer 3

像這樣的東西也可以，但我懷疑有更乾淨的方法。

class Merger<T> 
{ 
    Subject<T> _merged = new Subject<T>(); 

    public IObservable<T> Merged { get { return _merged; } } 

    public IDisposable Add(IObservable<T> newStream) 
    { 
     return newStream.Subscribe(_merged); 
    } 
} 

要刪除合併的流的東西，了IDisposable處置。