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 - 권장 읽음)에서 볼 수있는 'HOT'관측 가능 사례입니다.

Shlomo가 말했듯이, 주제는 Rx 세계의 "가변 변수"이므로 Obsevable.create (이벤트를 듣고 그 결과에 따라 방출을 생성 함)를 사용하여 '차갑고'관찰 가능하게 만들 수 있습니다. 그런 다음 ConnectableObservable (share, publish)으로 변환하는 연산자를 사용하여 서로 다른 시간에 구독하는 여러 관찰자에게 멀티 캐스트 할 수 있도록 '핫'하게 만듭니다.

그러나이 경우 생산자가 클래스에 로컬인데 (이벤트는이 클래스에서 생성됩니다), 클래스 자체가 배출원이되므로이 목적으로 제목을 사용하는 것이 좋습니다. 이벤트를 생성하는 상호/상태 변수. (answer given by Erik Meijer 및이 blog post 기준)