android - 使用 Observable Zip 行为不当

Question

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

• A - 取件
• B - 交付

订单分为三种类型。 仅限自取、仅限送货 和自取和送货。在任何情况下，Pickups 都需要在 Deliveries 之前运行。 Delivery 只有 Pickup 标记为 true。仅限Pickup，需要在关闭时pickup up and delivery。这就是为什么我需要 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);
}

zipper

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

利用 zipper

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

   之前/li>

2. 通读 Rx Documentation for Zip我可以看到它不会像我预期的那样工作，因为所有 getPickedOrders() 在 getDeliveredOrders() 运行之前首先运行。它必须一项一项地完成。例如:一次取货，然后一次送货

如果您能帮助我们理解正在发生的事情，我们将不胜感激。谢谢

Answer 1

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

• 仅取件:需要运行取件流程，然后他们完成。
• 仅交付:需要运行交付流程，然后他们才完成。
• Pickup and Delivery:需要先通过 Pickup，然后通过 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());