java - 收听然后发送订阅 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

I don't want to send the message then listen because, if the thread gets suspended or the response is super fast, I could miss it.

使用主题。 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();
}

通过这种方式您可以发送消息，每当您准备好收听时，您将收到最新发送的消息