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


A-取件
B-送货


订单有三种。 Pickup OnlyDelivery OnlyPickup And Delivery。在每种情况下,Pickups必须先运行DeliveriesDelivery仅已经将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());

10-08 05:33
查看更多