Rxjava: How to drop some object (temporary disconnect) when there is pending operation.

[midnight-wonderer]

Hello I have a question.
Sometimes I want to send an http request when a button is pressed but I don't want to send too much http request I just want to drop any button press emission when there is a pending http request.

I attempt with a couple solution with some flaw.

Attempt 1
_Solution:_ use onBackpressureDrop and use reactive pull Subscriber as parameter to subscribe().
_Issue:_ This solution require some code implementation in subscriber instance the problem is
onBackpressureDrop and subscriber code is too far away from each other (different file).
Make it hard to reasoning why the Subscriber instance implement the way it is. This approach will be perfect if I can move reactive pull up the chain.
e.g.

Observable buttonPress;
Func1 sendHttpRequest;
buttonPress.onBackpressureDrop().flatMap(sendHttpRequest).[ *** request(1) Operation *** ]

Basically the same as reactive pull but instead of doing it in subscriber do it with some operator inside transformation chain. I simply don't know such method (or operator) if you know how can I archive the same effect please tell me.

Attempt 2
_Solution:_ use ConnectableObservable to temporary disconnect flatMap operation from source.
_Issue:_ So imparative and the actual code looks bad break the purpose of using reactive paradigm.
I might did it the wrong way if you have an example to bring this idea into practice I will be very glad to see your example.

If you think both of my attempt are not even close to simple implementation available out of the box. I will be happy to see your suggestion on the original issue.

[akarnokd]

I don't think you should over complicate this. When you press the button, it should disable itself and once the background process terminates normally or with error, re-enable the button.

[midnight-wonderer]

Thank you @akarnokd for your kind reply.
It may looks over complicate but I try not to rely on side effects.

For solution based on side effect I found that simply use boolean field is better than disable the button (developer mental wise) because the boolean field is much closer to where it is required to be used so anyone can reasoning about why it is there without navigating around.

Both button disabling and boolean field can be an option if there is no other way around.
I just curious if there is solution based on observable chaining alone without too much influence from side effects. Plus I feel like I will understand more of RxJava this way.

[abersnaze]

Check out debounce() there are three flavors.

[midnight-wonderer]

Hi @abersnaze thank you for the suggestion.
I never figured out what that overload of debounce do.
Obviously in my use case debounce with time duration parameter is out of question.
That left only the mysterious one.

When I look at the calling signature I hope it work the same as flatMap with emission dropping feature but it is not.

Let says I do this

debounce(new Func1<Long, Observable<String>>() {
  @Override public Observable<String> call(Long oldValue) {
    return performStringRequest();
  }
})

It will result in Observable<Long> with collapsing element remove? correct?
I just wonder how can I use the result of performStringRequest() here
since the result of debounce is still Observable<Long> not Observable<String>.

Could you elaborate how this works?

It actually seems like a bug? (I mean just weird operator?) to me made me curious what was it used for.

[abersnaze]

The values from the Observable<U> are dropped only the completion of that observable is used to indicate when more Ts are propagated.

[midnight-wonderer]

After some time I kind of understand the functional of those debounce overload it seems to interest only onComplete signal but I still curious about real world usage of it.
It seems not a viable option for my use case though.

Do anyone have real world scenario about this operation?

Anyway I do some more research and found out that flatmap have maxConcurrent overload parameter.
The question then boiled down to:
If I set maxConcurrent to 1 will it generate backpressure that can be drop with onBackpressureDrop() or not?

So what I am trying to achieve will be as simple as

Observable buttonPress;
Func1 sendHttpRequest;
buttonPress.onBackpressureDrop().flatMap(sendHttpRequest, 1)

[abersnaze]

Here is another idea

        AtomicBoolean active = new AtomicBoolean(false);
        interval(100, TimeUnit.MILLISECONDS).filter(t -> {
            System.err.println("t=" + t);
            return !active.get();
        }).flatMap(t -> {
            active.set(true);
            return just("t=" + t + " completed").delaySubscription(1000, TimeUnit.MILLISECONDS).doOnUnsubscribe(() -> active.set(false));
        }).toBlocking().forEach(System.out::println);

produces the output

t=0
t=1
t=2
t=3
t=4
t=5
t=6
t=7
t=8
t=9
t=10
t=0 completed
t=11
t=12
t=13
t=14
t=15
t=16
t=17
t=18
t=19
t=20
t=21
t=11 completed
t=22

[abersnaze]

The real world use for debounce is interactive search box where you don't want every key press to cause a http call to a search service.

onTextBoxChange().debounce(100 ms).switchMap(text -> search(text)).subscribe(updateResults)

I'm thinking the reason it doesn't work for you is that I'm assuming you don't want to cancel an in flight request if the user give you more input because the operation is not idempotent.

[midnight-wonderer]

@abersnaze nice example although it is relying on side effect of boolean as I mentioned the implementation is still better than mine.
Regarding the real world scenario I referred to those particular variant of debounce where there is no time period as parameter.

[midnight-wonderer]

Just update to anyone with the same issue.
The solution I described seems to work.
If you want to see it for yourself here simple demo code:

    Observable
      .interval(500, TimeUnit.MILLISECONDS)
      .onBackpressureDrop(new Action1<Long>() {
        @Override public void call(Long aLong) {
          // Timber is just a logging library you can replace with any logging machanism
          Timber.d("dropped");
        }
      })
      .flatMap(new Func1<Long, Observable<Void>>() {
        @Override public Observable<Void> call(Long aLong) {
          Timber.d("long operation start");
          return Observable.<Void>just(null).delay(3200, TimeUnit.MILLISECONDS);
        }
      }, 1)
      .subscribe(new Action1<Void>() {
        @Override public void call(Void aVoid) {
          Timber.d("long operation completed");
        }
      });