Gibt es einen Fall, dass ein Beobachter von einigen Observablen in RxJava geteilt wird?

Question

Ich denke, dass es falsch ist, eine Beobachterinstanz mit einigen Observablen zu teilen. Zum Beispiel

Observable o1 = Observable.interval(1, TimeUnit.MILLISECOND); 
Observable o2 = Observable.interval(1, TimeUnit.MILLISECOND); 

Observer observer = new Observer() { ... } 

o1.subscribe(observer); 
o2.subscribe(observer); 

Seit der OnCompleted des Beobachters() würde einige Male aufgerufen werden, könnte es nicht akzeptabel sein.

Aber ich konnte keine Dokumentation darüber finden. Also frage ich mich, ob es einen Fall gibt, einen Beobachter mit einigen Observablen zu teilen.

Answer 1

Durch den Entwurf werden die Beobachter einmal erstellt, dass Sie die Observable abonnieren, und sobald Ihr Element emittiert wird, wird Ihr Beobachter abbestellen und entfernen.

Vermeiden Sie die Verwendung von Observables und verwenden Sie Relay, das Beobachter nie austragen wird, wenn es abgeschlossen wird, da onComplete-Rückruf nicht bereitgestellt wird.

/** 
* ReplayRelay it works just like hot observables, once that an observer subscribe, the Relay will replay all items already emitted 
* to another observer. 
* it should return 1,2,3,4,5 for both observers. 
*/ 
@Test 
public void testReplaytRelaySharingObserver() { 
    ReplayRelay<String> relay = ReplayRelay.create(); 
    Subscription subscription = relay.subscribe(result -> System.out.println("Observer1:" + result)); 
    relay.call("1"); 
    relay.call("2"); 
    relay.call("3"); 
    relay.subscribe((Observer<? super String>) subscription); 
    relay.call("4"); 
    relay.call("5"); 
} 

Sie können für Ihre Informationen https://github.com/politrons/reactive/blob/master/src/test/java/rx/relay/Relay.java