Слушайте, а затем отправляйте по подписке с RxJava

Question

Я продолжаю работать в ситуациях, когда хочу, чтобы я мог услышать ответ на запрос через RxJava. Проблема в том, что я не уверен, как настроить Observable, чтобы я слушал события и отправлял сообщение на подписку в правильном порядке. Я не хочу посылать сообщение, а потом слушать, потому что, если поток приостанавливается или ответ очень быстрый, я могу его пропустить. Это ближе всего я мог думать о моей собственной

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

или

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

Но это все еще кажется, что я все еще мог послать сообщение и не слушая ответа. Кто-нибудь еще пытался это сделать? Я чувствую, что мне действительно нужен вид afterCreate().

Answer 1

Я не хочу отправлять сообщение, а затем слушать, потому что, если нить приостанавливается или ответ очень быстрый, я могу пропустить его.

Использовать Subject. Либо BehaviorSubject (испускает всегда последнее наблюдаемое излучаемое новому подписчику), либо ReplySubject (испускает все Observable, исходящие от нового абонента). Я не уверен, о всей логике, но вы могли бы иметь что-то вроде:

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(); 
} 

Таким образом, вы можете отправить сообщение и всякий раз, когда вы будете готовы слушать юй получит, в этом случае, последнее сообщение было отправлено

Answer 2

doOnSubscribe - это именно то, что вам нужно, это «afterCreate», который вы ищете. В первом примере первое сообщение будет отправлено после подписки, и в это время Observable уже готов обработать ответ.

Что касается вашего вопроса - Кто-нибудь еще пытался это сделать? - ответ - да. Я использую ту же технику, что и ваш первый пример, чтобы развернуть часть кода.