Hören Sie dann senden Sie auf abonnieren mit RxJava

Question

Ich laufe immer in Situationen, in denen ich auf eine Antwort auf eine Anfrage über RxJava hören wollen. Das Problem ist, ich bin mir nicht sicher, wie ich das Observable so einrichten soll, dass ich auf Ereignisse lausche und die Nachricht auf subscribe in der richtigen Reihenfolge sende. Ich möchte die Nachricht nicht senden und dann zuhören, denn wenn der Thread ausgesetzt wird oder die Antwort sehr schnell ist, könnte ich es vermissen. Dies ist in der Nähe von was ich denken konnte auf meinem eigenen

connection.onReceivedMessage() 
      .doOnSubscribe(() -> connection.send(message)) 
      .filter(message -> message.id == id) 
      ... // do stuff 

oder

Observable.defer(() -> { 
    connection.send(message); 
    return connection.onReceivedMessage(); 
})... // do stuff 

Aber diese scheinen immer noch wie ich nach wie vor die Nachricht senden konnte und nicht für die Antwort zuhören. Hat jemand anderes versucht, dies zu tun? Ich habe das Gefühl, ich möchte wirklich eine Art AfterCreate().

Answer 1

Ich will nicht die Nachricht dann senden hören, weil, wenn der Thread ausgesetzt wird oder die Antwort ist super schnell, ich es vermissen könnte.

Verwenden Sie eine Subject. Entweder eine BehaviorSubject (emittiert immer die neueste beobachtbare emittiert zu neuen Abonnenten) oder eine ReplySubject (sendet alle Observable emittiert zu neuen Abonnenten). Ich bin über die ganze Logik nicht sicher, aber man könnte so etwas wie haben:

public BehaviorSubject mMessageBehaviorSubject = BehaviorSubject.create(); 


private void sendMessage() { 
    connection.onReceivedMessage() 
      .doOnSubscribe(() -> connection.send(message)) 
      .filter(message -> message.id == id) 
      .subscribe(mSubject::onNext, Throwable::printStackTrace); 
} 


public Observable<String> getMessageObservable() { 
    return mMessageBehaviorSubject.asObservable(); 
} 

diese Weise können Sie die Nachricht senden könnten und wenn Sie bereit sind zu hören yu erhalten, in diesem Fall gesendet letzte Nachricht

Answer 2

doOnSubscribe ist genau das, was Sie brauchen, es ist das "afterCreate", nach dem Sie suchen. In Ihrem ersten Beispiel wird die erste Nachricht nach dem Abonnement gesendet. Zu diesem Zeitpunkt ist das Observable bereits bereit, die Antwort zu verarbeiten.

Wie für Ihre Frage - Hat jemand anderes versucht, dies zu tun? - Die Antwort ist ja. Ich benutze die gleiche Technik wie Ihr erstes Beispiel, um ein Stück Code zu ruxifizieren.