Rxで観測可能な状態を維持する最適な方法

Question

観測可能な状態を維持するために、より良いRxJavaパターンが何であるかについて質問があります。

シンプルに保つために、システム内で特定の状態（単純なブール値のフラグと見なします）を追跡し、それを観測可能な方法で公開する必要があるStateManagerクラスがあるとします。このマネージャーは非常に長いライフサイクルを持っている

class StateManager { 
    Observable<Boolean> state(); 
    ... 
} 

を、いつでも購読または購読解除う複数の「クライアント」（例えばビュー、他の経営者など）を持つことができる：このように、それは、以下のような方法を持っています。状態は、内部イベントに基づいて変更されます。

これに対処するための最も明白な方法は、消費者が直接フックするためにどのBehaviourSubject状態を保持するために、次のようになります。

class StateManager { 

    Subject mStateSubject = BehaviourSubject.create(true);   

    Observable<Boolean> state() { 
     return mStateSubject.asObservable(); 
    }  
    ... 
} 

これまでより良い方法はありますか？

Answer 1

Subjectsはおそらく反応性ライブラリーを使用する最も望ましい方法ですが、確かに有効です。

機能リアクティブプログラミングは、状態なしで最も効果的です。 Subjectsは状態の一種です。 Observableが関数演算子の組み合わせとして定義されるようにコードを変更することをお勧めします。これにより、観測対象が発信しているメッセージを簡単にテストして管理することができます。

---

私はもっとC＃の開発者ですから、別の構文を許して欲しいと思います。ここでは例です：C＃の土地で

void Main() 
{ 
    var tracker = new AddTracker(); 
    tracker.getSums().Subscribe(i => Console.WriteLine(i)); 
    Observable.Interval(TimeSpan.FromMilliseconds(100)) 
     .Timestamp() 
     .Select(t => t.Timestamp.Second) 
     .Take(20) 
     .Subscribe(i => tracker.setA(i % 7)); 

    Observable.Interval(TimeSpan.FromMilliseconds(75)) 
     .Timestamp() 
     .Select(t => t.Timestamp.Millisecond) 
     .Take(30) 
     .Subscribe(i => tracker.setB(i % 9)); 

} 

public class AddTracker 
{ 
    private readonly ISubject<int> _a; 
    private readonly ISubject<int> _b; 
    private readonly IObservable<int> _sums; 
    private readonly IDisposable _dummySub; 

    public AddTracker() 
    { 
     _a = new BehaviorSubject<int>(0); 
     _b = new BehaviorSubject<int>(0); 
     _sums = _a 
      .CombineLatest(_b, (a, b) => a + b) 
      .Replay(1) 
      .RefCount(); 
     _dummySub = _sums.Subscribe(_ => { }); 
    } 

    public void setA(int value) 
    { 
     _a.OnNext(value); 
    } 

    public void setB(int value) 
    { 
     _b.OnNext(value); 
    } 

    public IObservable<int> getSums() 
    { 
     return _sums; 
    } 
} 

、あなたは軽度の改善である、イベントのための_aと_b科目を入れ替えることができます。私はJavaにはファーストクラスのイベントはないことを知っていますので、それが翻訳されるかどうかはわかりません。

しかし、基本的にC＃とJavaの両方で、質問する必要があるのは... setAとsetBの呼び出しの原因は何ですか？そして、あなたはこれでそれらを置き換えることができます：

void Main() 
{ 
    var aStream = Observable.Interval(TimeSpan.FromMilliseconds(100)) 
     .Timestamp() 
     .Select(t => t.Timestamp.Second) 
     .Take(20); 

    var bStream = Observable.Interval(TimeSpan.FromMilliseconds(75)) 
     .Timestamp() 
     .Select(t => t.Timestamp.Millisecond) 
     .Take(30); 

    var tracker = new AddTracker(aStream, bStream); 
    tracker.getSums().Subscribe(i => Console.WriteLine(i)); 

} 

public class AddTracker 
{ 
    private readonly IObservable<int> _sums; 
    private readonly IDisposable _dummySub; 

    public AddTracker(IObservable<int> a, IObservable<int> b) 
    { 
     _sums = a 
      .CombineLatest(b, (aItem, bItem) => (aItem % 9) + (bItem % 7)) 
      .Replay(1) 
      .RefCount(); 
     _dummySub = _sums.Subscribe(_ => {}); 
    } 

    public IObservable<int> getSums() 
    { 
     return _sums; 
    } 
} 

要するに、被験者から始める必要があります。その後、あなたの主題を取って、あなたのロジックからできるだけ遠くに押し続けてください。

Answer 2

上記のケースは、「ホット」と呼ばれるもので、サブスクリプション外で作成されたプロデューサ（排出源）が （Ben Lesh：Hot vs Cold Observables - 推奨読書）に記載されています。

Shlomoが言っているように、被験者はRxワールドの「可変変数」であり、Obsevable.create（イベントを聴いてそれらに基づいてエミッションを生成する）を使用することで、別の時間にサブスクライブする複数のオブザーバーにマルチキャストするために、ConnectableObservable（共有、パブリッシュなど）に変換する演算子を使用して「ホット」にします。

しかし、この場合、プロデューサはあなたのクラスに対してローカルです（イベントはこのクラスによって生成されます）。クラス自体が排出源であり、この目的のためにSubjectを使用することは問題ありません。イベントを生成する相互/状態変数 （answer given by Erik Meijerに基づき、これはblog post）