RxJava teilen sich eine Emissionen des beobachtbaren zwischen mehreren Teilnehmern

Question

Ich habe folgendes Problem:

Ich habe einen beobachtbaren, dass einige Arbeit zu tun, aber auch andere Observablen brauchen die Ausgabe dieses beobachtbaren zu arbeiten. Ich habe versucht, mehrere Male auf dem gleichen Observablen zu subskribieren, aber innerhalb des Logs sehe ich, dass das ursprüngliche Observable mehrfach initiiert wird.

das ist meine beobachtbaren das ist Erstellen Sie das Objekt:

Observable.create((Observable.OnSubscribe<Api>) subscriber -> { 
      if (mApi == null) { 
       //do some work 
      } 
      subscriber.onNext(mApi); 
      subscriber.unsubscribe(); 
     }) 

das ist meine beobachtbar, dass das Objekt

loadApi().flatMap(api -> api....())); 

Ich verwende

.subscribeOn(Schedulers.io()) observable.observeOn(AndroidSchedulers.mainThread()) 
       .unsubscribeOn(Schedulers.io() 

auf allen Observablen muss.

Answer 1

Ich bin mir nicht sicher, ob ich Ihre Frage richtig verstanden habe, aber ich denke, Sie suchen nach einer Möglichkeit, die Emissionen eines Observablen zwischen mehreren Teilnehmern zu teilen. Es gibt mehrere Möglichkeiten, dies zu tun. Zum einen könnte man ein Connectable Observable wie so verwenden:

ConnectableObservable<Integer> obs = Observable.range(1,3).publish(); 
obs.subscribe(item -> System.out.println("Sub A: " + item)); 
obs.subscribe(item -> System.out.println("Sub B: " + item)); 
obs.connect(); //Now the source observable starts emitting items 

Ausgang:

Sub A: 1 
Sub B: 1 
Sub A: 2 
Sub B: 2 
Sub A: 3 
Sub B: 3 

Alternativ können Sie verwenden, um eine PublishSubject:

PublishSubject<Integer> subject = PublishSubject.create(); //Create a publish subject 
subject.subscribe(item -> System.out.println("Sub A: " + item)); //Subscribe both subscribers on the publish subject 
subject.subscribe(item -> System.out.println("Sub B: " + item));  
Observable.range(1,3).subscribe(subject); //Subscribe the subject on the source observable 

Ausgang:

Sub A: 1 
Sub B: 1 
Sub A: 2 
Sub B: 2 
Sub A: 3 
Sub B: 3 

Beide o f Diese Beispiele sind single threaded, aber Sie können die Aufrufe von observeOn oder subscirbeOn einfach hinzufügen, um sie asynchron zu machen.

Answer 2

Zunächst einmal mit Observable.create ist schwierig und leicht zu falsch. Sie brauchen so etwas wie

Observable.create(subscriber -> { 
     if (mApi == null) { 
      //do some work 
     } 
     if (!subscriber.isUnsubscribed()) { 
      subscriber.onNext(mApi); 
      subscriber.onCompleted(); 
      // Not subscriber.unsubscribe(); 
     }    
}) 

Sie

ConnectableObservable<Integer> obs = Observable.just(1).replay(1).autoConnect(); 

alle nachfolgenden Teilnehmer sollten die einzelnen ausgesandten Artikel

obs.subscribe(item -> System.out.println("Sub 1 " + item)); 
obs.subscribe(item -> System.out.println("Sub 2 " + item)); 
obs.subscribe(item -> System.out.println("Sub 3 " + item)); 
obs.subscribe(item -> System.out.println("Sub 4 " + item));