Ituoke Ajanlekoko Ituoke Ajanlekoko - 1 month ago 21
Android Question

Using Observable Zip misbehaving

I have two observables(A, B), and I want the first to finish running before the second runs. But, that's not even the problem I'm having. The problem is that, when A is added before B, B doesn't run at all unless I place B before A then, the two runs. But, the scenario I'm in is like thus:


  • A - Pickup

  • B - Delivery



There are three types of orders.
Pickup Only
,
Delivery Only
and
Pickup And Delivery
.
Pickups
need to run before
Deliveries
in every situation. A
Delivery
only already have
Pickup
marked as
true
. A
Pickup
only, needs to be
picked up and delivered
on it being closed. Which is why I need
Pickup
to send all locally saved pickups first before sending deliveries. So, I did this:

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


Delivery

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


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


Problem


  1. I don't know why
    getDeliveredOrders()
    doesn't get called unless I move it to the first before
    getPickedOrders()

  2. Reading through Rx Documentation for Zip I can see that it's not going to work as I expected where all of
    getPickedOrders()
    runs first before
    getDeliveredOrders()
    runs. It'll have to do it one by one. E.g: One of Pickup and then One of Delivery



Any help to understand what's going on would be appreciated. Thanks

Answer

Ok, so if I got that right:

  • Pickup only: need to run through the Pickup process, then they complete.
  • Delivery only: need to run through the Delivery process, then they complete.
  • Pickup and Delivery: need to run through Pickup first, then through Delivery.

On a very high level, almost preudo-code, why does this process not work?

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

If you have different sources for the three types:

Observable<Item> items = Observable.merge(
     pickupSource(),
     deliverySource(),
     pickupAndDeliverySource());