我有两个可观察值(A,B),我希望第一个可在第二个可运行之前完成运行。但是,这甚至不是我遇到的问题。问题是,当在B之前添加A时,B根本不会运行,除非我将B放在A之前,这两个都将运行。但是,我所处的场景是这样的:
A-取件
B-送货
订单有三种。 Pickup Only
,Delivery Only
和Pickup And Delivery
。在每种情况下,Pickups
必须先运行Deliveries
。 Delivery
仅已经将Pickup
标记为true
。仅Pickup
,在关闭时需要为picked up and delivered
。这就是为什么我需要Pickup
首先发送所有本地保存的提货,然后再发送交货。因此,我这样做:
拾起
private Observable<UpdateMainResponse> getDeliveredOrders() {
String token = PrefUtil.getToken(context);
BehaviorSubject<Integer> pageControl = BehaviorSubject.create(1);
Observable<UpdateMainResponse> ret = pageControl.asObservable().concatMap(integer -> {
if (integer - 1 != deliveryUpdate.size()) {
Log.e(TAG, "DeliveredOrders: " + deliveryUpdate.size());
RealmOrderUpdate theDel = deliveryUpdate.get(integer-1);
Log.e(TAG, "DeliveryUpdate: " + theDel.toString());
DeliverOrder pickupOrder = new DeliverOrder();
pickupOrder.setUuid(theDel.getUuid());
pickupOrder.setCode(theDel.getDest_code());
pickupOrder.setDelivered_lat(theDel.getLoc_lat());
pickupOrder.setDelivered_long(theDel.getLoc_long());
return apiService.deliverOrder(theDel.getOrderId(), token, pickupOrder)
.subscribeOn(Schedulers.immediate())
.doOnNext(updateMainResponse -> {
try {
Log.e(TAG, updateMainResponse.toString());
realm.executeTransaction(realm1 -> theDel.deleteFromRealm());
} catch (Exception e) {
e.printStackTrace();
} finally {
pageControl.onNext(integer + 1);
}
});
} else {
return Observable.<UpdateMainResponse>empty().doOnCompleted(pageControl::onCompleted);
}
});
return Observable.defer(() -> ret);
}
交货
private Observable<UpdateMainResponse> getPickedOrders() {
Log.e(TAG, "PickedOrders: " + pickUpdate.size());
String token = PrefUtil.getToken(context);
BehaviorSubject<Integer> pageControl = BehaviorSubject.create(1);
Observable<UpdateMainResponse> ret = pageControl.asObservable().concatMap(integer -> {
Log.e(TAG, "MainPickedInteger: " + integer);
if (integer - 1 != pickUpdate.size()) {
RealmOrderUpdate thePick = pickUpdate.get(integer - 1);
Log.e(TAG, "PickedUpdate: " + thePick.toString());
PickupOrder pickupOrder = new PickupOrder();
pickupOrder.setUuid(thePick.getUuid());
pickupOrder.setCode(thePick.getSource_code());
pickupOrder.setPicked_lat(thePick.getLoc_lat());
pickupOrder.setPicked_long(thePick.getLoc_long());
return apiService.pickupOrder(thePick.getOrderId(), token, pickupOrder)
.subscribeOn(Schedulers.immediate())
.doOnNext(updateMainResponse -> {
try {
Log.e(TAG, updateMainResponse.toString());
realm.executeTransaction(realm1 -> thePick.deleteFromRealm());
} catch (Exception e) {
e.printStackTrace();
} finally {
pageControl.onNext(integer + 1);
}
});
} else {
return Observable.<UpdateMainResponse>empty().doOnCompleted(pageControl::onCompleted);
}
});
return Observable.defer(() -> ret);
}
拉链
private Observable<ZipperResponse> batchedZip() {
return Observable.zip(getPickedOrders(), getDeliveredOrders(), (updateMainResponse, updateMainResponse2) -> {
List<UpdateMainResponse> orders = new ArrayList<>();
bakeries.add(updateMainResponse);
bakeries.add(updateMainResponse2);
return new ZipperResponse(orders);
});
}
利用拉链
public void generalUpload(APIRequestListener listener) {
batchedZip.subscribe(new Subscriber<ZipperResponse>() {
@Override
public void onCompleted() {
listener.didComplete();
unsubscribe();
}
@Override
public void onError(Throwable e) {
listener.handleDefaultError(e);
unsubscribe();
}
@Override
public void onNext(ZipperResponse zipperResponse) {
Log.e(TAG, zipperResponse.size());
}
});
}
问题
我不知道为什么
getDeliveredOrders()
不会被调用,除非我将其移到getPickedOrders()
之前的第一个通读Rx Documentation for Zip,我可以看到它无法正常运行,正如我预期的那样,所有
getPickedOrders()
首先在getDeliveredOrders()
运行之前运行。它必须一个接一个地做。例如:取件之一,然后交货任何帮助了解正在发生的事情将不胜感激。谢谢
最佳答案
好吧,所以如果我说对了:
仅取件:需要完成取件过程,然后完成。
仅交付:需要完成交付过程,然后完成。
取件和交付:需要先通过取件运行,然后再通过交付。
在非常高级的几乎是伪代码的情况下,为什么此过程不起作用?
Observable<Item> performPickup(Item item);
Observable<Item> performDelivery(Item item);
Observable<Items> items = ...;
items
.flatMap(item -> item.needsPickup() ? performPickup(item) : Observable.just(item))
.flatMap(item -> item.needsDelivery() ? performDelivery(item) : Observable.just(item))
.doOnNext(completedItem -> ...)
如果这三种类型的来源不同:
Observable<Item> items = Observable.merge(
pickupSource(),
deliverySource(),
pickupAndDeliverySource());