Put in / remove from the basket using Rx

Question

I have two streams of data:

1. Boolean stream, which indicates whether the item exists in the basket.
2. Clicks stream, which initiates item put/delete into/from the basket.

Firstly, I want to change caption of the toggle button according to boolean stream. It's simple. Secondly, I want to combine the latest value from the Boolean stream and event from Clicks stream in order to initiate either put or delete request.

Here's what I've tried so far:

// handling the toggle button caption
inBasketStream.subscribe(inBasket -> {
    mPurchaseButton.setText(inBasket ? "Already in basket" : "Purchase");
});

// handling clicks and deciding whether to add or remove item
Observable.combineLatest(
        inBasketStream,
        ViewObservable.clicks(mPurchaseButton),
        (inBasket, onClickEvent) -> inBasket).subscribe(inBasket -> {
            if (inBasket) {
                mRemoveFromBasket.call(mItemId);
            } else {
                mAddInBasket.call(mItemId);
            }
        }
);

However, combineLatest doesn't do my job. Instead it results in recursion since at least one clicks event occured and inBasketStream is updated on operation completion. Zip won't help much as well since in basket stream might be changed from another place and there's, hence, a scenario for stacking multiple boolean values, which will make zip take obsolete(just the next one, while there could be more values already stacked upon) value on next click event.

So, I need a way to get the last value from the boolean stream and use it whenever click event happens. Of course, I could use a boolean field for that (subscribe to inBasketStream and update it onNext) but I don't feel that it complies with functional and there has to be a composition to handle this issue without additional fields/variables.

Thanks.

P.S. both observables are hot

Answer 1

Take a look at withLatestFrom. It will only propagate when the primary source emits while combining the latest values from the other sources.

ViewObservable.clicks(mPurchaseButton)
              .withLatestFrom(inBasketStream, 
                             (dontCare, inBasket) -> inBasket)
.subscribe(/*do your update*/);