Pausing and resuming a stream in RxJava

Introducing the valve extension operator.

A water valve is an accurate analogy.

I’ve spent the last couple of days trying to achieve something I thought would be fairly simple - pausing an active stream, buffering any incoming upstream items and then emitting them all once the stream is resumed.

I figured that there would be an existing operator that already did this but having dug through the documentation it appears I was wrong. There are operators that come close, notably buffer and window, however these are only designed to emit items in batches rather than a stream, which isn’t what I wanted.

Eventually I decided to post a request on the RxJava repository where the lovely David Karnok decided that while he saw the value, he thought it would be better suited to his RxJava2Extensions library rather than the main RxJava repo. There’s a lot of nice features in there but if you’re only interested in the pause and resume functionality then you might be best just pulling out that part of the code.

FlowableValve

David created an operator called FlowableValve that does precisely what I described, you pass in a Publisher that emits a boolean, if it emits true then the valve opens, if it emits false the valve closes. You can also specify whether you want the valve to be open or closed by default and the size of the buffer. The underlying implementation simply queues items when the valve is closed and emits them all once the valve is opened.

So let’s see some code.

First we need to create our Publisher that acts as the switch, opening and closing the valve, for this I’ve gone with a PublishProcessor like so:

PublishProcessor<Boolean> valve = PublishProcessor.create();

Next we create a new stream that outputs a sequential value every second and writes it to the log which will make it easier to know when the stream is paused as the logs won’t be output.

Flowable.interval(1, TimeUnit.SECONDS)
        .compose(FlowableTransformers.valve(valve, true))
        .subscribe(new Consumer<Long>() {
            @Override
            public void accept(final Long count) throws Exception {
                Log.d("Count", "Count = " + count);
            }
        });

The magic here is the compose method, it allows us to pass in a function that transforms the original Publisher (which in this case is a Flowable). Here we pass in the FlowableTransformers.valve() function contained within the RxJava2Extensions, giving it the valve Publisher we created earlier and a default value of true, meaning the valve is open by default.

If you run this code you will get an output similar to this:

You should get a new incremented number every second.

Now we want to be able to open and close the valve when we want, for this we’ll use a button to keep things simple, I’m using the RxBinding library to make it easier to create reactive sequences from Views but you can just as easily do without it.

private boolean isValveOpen = true;
RxView.clicks(button)
        .subscribe(new Action1<Void>() {
            @Override
            public void call(final Void aVoid) {
                isValveOpen = !isValveOpen;
                Log.d("Count", "Valve is " + (isValveOpen ? "open." : "closed."));
                valve.onNext(isValveOpen);
            }
        }));

Every time the button is clicked we toggle the value of isValveOpen and then pass the new value to any subscribers via onNext(), this updates the valve method and either opens or closes the valve depending on the value of isValveOpen.

Now you should see something like this:

When the valve is closed the items are buffered, once the valve is opened again all of the buffered values are emitted and the stream continues from where it left off.

This is just a simple example of what you can use this operator for, I’m planning on using it to achieve something quite powerful in my next app which I’ll share with you all once it’s in place, stay tuned!