Zip可观察对象列表到另一个Zip可观察对象RxJava2

Question

我正在尝试zipzipObservables的列表，但问题是我每次只能从压缩的可观察对象中获得相同的值。我这样做的原因是要从ble执行两个操作1st reading index和2nd reading data一定次数-下面的示例是6次。

不知道如何处理这个与RxJava2

下面是代码片段

private Observable<Pair<byte[],byte[]>> getValueFromIndication(RxBleConnection rxBleConnection){
final PublishSubject<Boolean> unsubscribeSubject = PublishSubject.create();

return Observable.zip(
rxBleConnection.setupIndication(Data.INDEX,NotificationSetupMode.QUICK_SETUP).flatMap(it->it).takeUntil(unsubscribeSubject),
rxBleConnection.setupIndication(Data.DATA,NotificationSetupMode.QUICK_SETUP).flatMap(it->it).takeUntil(unsubscribeSubject),
(bytes, bytes2) -> {
unsubscribeSubject.onNext(true);
return Pair.create(bytes,bytes2);
}
);
}

从我的主流，我首先创建列表的Observables和压缩它，并传递它

.flatMap(rxBleConnection -> {

List<Observable<Pair<byte[],byte[]>>> observableList = new ArrayList<>();
for(int i=0;i<6;i++){
//Creating list of observables so that 6 times this function gets fire
observableList.add(getValueFromIndication(rxBleConnection));
}
// Zipping Zipped list of observables 
return Observable.zip(observableList,Data::OperationReadings);
}).subscribe(bytes->{

})

在这里，我总是在Data::OperationReadings中得到相同的值。目前，我得到了以下我不想要的数据:

每次索引和值相同

INDEX [1] DATA [10,30,20,30,33,0]
INDEX [1] DATA [10,30,20,30,33,0]
INDEX [1] DATA [10,30,20,30,33,0]
INDEX [1] DATA [10,30,20,30,33,0]
INDEX [1] DATA [10,30,20,30,33,0]
INDEX [1] DATA [10,30,20,30,33,0]

期望的数据如下

每次索引和值不同

INDEX [1] DATA [10,30,20,30,33,0]
INDEX [2] DATA [11,11,2,0,3,0]
INDEX [3] DATA [0,0,0,0,33,0]
INDEX [4] DATA [10,30,0,30,3,0]
INDEX [5] DATA [10,0,0,30,3,0]
INDEX [6] DATA [10,0,20,30,3,9]

Answer 1

重复6次获得相同数据的原因是您同时订阅了单个getValueFromIndication()。实际上，每个Observable都是并行运行的。您希望按顺序运行每个订阅。解决方案可能是替换以下内容:

return Observable.zip(observableList,Data::OperationReadings);

:

return Observable.concat(observableList) // we want to subscribe each Observable from list after the previous one will complete
.toList() // we want to gather all results from individual Observables from the list — this returns a Single
.toObservable() // get back to the Observable class so the types will match
.map(Data::OperationReadings); // we map it into the OperationReadings class