使用可观察Zip错误行为

Question

我有两个可观察对象(A, B)，我希望第一个在第二个运行之前完成运行。但是，这甚至不是我遇到的问题。问题是，当A加在B前面时，B根本不会运行除非我把B放在A前面然后，这两次运行。但是，我所处的场景是这样的:

• A -拾取
• B - Delivery

订单有三种类型。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());
        }
    });
}

1. 我不知道为什么getDeliveredOrders()不被调用，除非我把它移动到getPickedOrders()之前

2. 阅读Zip的Rx文档，我可以看到它不会像我预期的那样工作，所有getPickedOrders()在getDeliveredOrders()运行之前首先运行。它必须一个接一个地做。例:一次取货，一次发货

如果你能帮助我理解发生了什么，我将不胜感激。由于

Answer 1

好的，如果我没弄错的话:

• 仅拾取:需要通过拾取过程运行，然后它们完成。
• Delivery only:需要通过Delivery流程运行，然后它们完成。
• 取货和发货:需要先取货，再送货。

在一个非常高的层次上，几乎是先验代码，为什么这个过程不起作用?

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