我正在使用在Android设备上运行的RxJava。我有2个PublishSubjects，它们定期发布通过调用OnNext（）从蓝牙特性接收的数据（字符串）。一个特征（A）确实存在，另一个（B）可能存在或不存在。我正在尝试从其中一个或两个中获取数据，并使用它们来做出一些决策。连接到蓝牙设备后以及第一次从这些特征接收数据时，我只想运行一次。

我真的很难使它正常工作。我可以使用CombineLatest（）或zip（）合并可观察对象，并使用take（）定义超时，但是我的逻辑似乎总是被调用两次。

例如

Observable.combineLatest(A,B.take(2,TimeUnit.SECONDS).defaultIfEmpty(""),(a,b) -> {
    //combine logic
    return aAndMaybeB;
}.observeon(AndroidSchedulers.mainThread()).firstOrError().subscribe( aAndMaybeB-> {
    //make some decisions
});

或

A.firstOrError().zipWith(B.take(2,TimeUnit.SECONDS).first(""),b) -> {
    //combine logic
    return aAndMaybeB;
}.observeon(AndroidSchedulers.mainThread()).subscribe( aAndMaybeB-> {
    //make some decisions
});

为什么组合逻辑和/或做出一些决策两次被调用？我怎么只能让它发生一次？

有很多线程详细介绍了如何组合可选的可观察对象，但我发现没有一个对象只讨论一次操作

Optional Observables in combineLatest

RxJava: combine two optional observables

RxJava: how to handle combineLatest() when one of the streams emits nothing

RxJava combine Observable with another optional Observable with timeout