wpf - マージされたシーケンスから IObservable を削除します

Question

Observable.Merge を使用して複数のシーケンスを結合し、UI の単一のビューに表示しています。ユーザーは、UI でシーケンス (フィード) を追加または削除することを選択できます。Merge を使用してフィードを結合しました。Merged シーケンスから IObservable を切り離す方法がわかりません。現在、必要なフィードを省略して、まったく新しい IObservable を作成しています。ViewModel が既にサブスクライブしている IObsevable を動的に追加および削除することは可能ですか?

Answer 1

using を見てIObservable<IObservable<T>>から、 を使用してくださいMerge。これにより、インナーを終了することでシーケンスを自動的に削除できますIObservable<T>。単純。

Answer 2

このようなものも機能しますが、もっときちんとした方法があると思います。

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 を破棄します。

Answer 3

これは開始するのに役立つかもしれませんが、再接続すると、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");

編集：

マージされた IO に別の 'c' を追加するには:

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