Limit actions with rxJava and retryWhen operator
<p>My application must do two things in general:</p>

<ul>
<li>Accept only one network request at the same time</li>
<li>Retry if request failed</li>
</ul>

<p>That's how I implement it:</p>

<pre><code>public class RequestsLocker {

    private volatile boolean isLocked;

    public &lt;T&gt; Observable.Transformer&lt;T, T&gt; applyLocker() {
        if(!isLocked()) {
            return observable -&gt; observable
                    .doOnSubscribe(() -&gt; {
                        lockChannel();
                    })
                    .doOnUnsubscribe(() -&gt; {
                        freeChannel();
                    });
        } else {
            return observable -&gt; Observable.error(new ChannelBusyException("Channel is busy now."));
        }
    }

    private void lockChannel() {
        isLocked = true;
    }

    private void freeChannel() {
        isLocked = false;
    }

    public boolean isLocked() {
        return isLocked;
    }

}
</code></pre>

<p>Looks nice.</p>

<p>Now my <code>retryWhen</code> implementation:</p>

<pre><code>public static Observable&lt;?&gt; retryWhenAnyIoExceptionWithDelay(Observable&lt;? extends Throwable&gt; observable) {
    return observable.flatMap(error -&gt; {
        // For IOExceptions, we  retry
        if (error instanceof IOException) {
            return Observable.timer(2, TimeUnit.SECONDS);
        }

        // For anything else, don't retry
        return Observable.error(error);
    });
}
</code></pre>

<p>There is how I use it:</p>

<pre><code>public Observable&lt;List&lt;QueueCarItem&gt;&gt; finishService(int id, PaymentType paymentType, String notes) {
    return carsQueueApi.finishService(id, new FinishCarServiceRequest(paymentType.getName(), notes))
            .compose(requestsLocker.applyLocker(RequestsLocker.RequestChannel.CHANGE));
}
</code></pre>

<p>...</p>

<pre><code>public void finishCarService(QueueCarItem carItem, PaymentType paymentType,
                             String notes, Subscriber&lt;List&lt;QueueCarItem&gt;&gt; subscriber) {
    queueApiMediator.finishService(carItem.getId(), paymentType, notes)
            .subscribeOn(ioScheduler)
            .observeOn(uiScheduler)
            .doOnError(this::handleError)
            .retryWhen(RxOperatorsHelpers::retryWhenAnyIoExceptionWithDelay)
            .subscribe(subscriber);
}
</code></pre>

<p>The main problem that <code>doOnUnsubscribe()</code> called on any error and then locker is open for any new request until the timer expires and resubscribing happens again. That's the problem. While the timer is ticking user can make another request.</p>

<p><strong>How I can fix it?</strong></p>