Code adapted to RX Completable not blocking onSubscribe thread

Question

I have some legacy, non-RX code that does some networking work by spawning a new Thread. When the work finishes, it invokes one method on a callback.

I don't have control over the thread this code runs on. It's legacy, and it spawns a new Thread by itself.

This can be simplified in something like:

interface Callback {
    void onSuccess();
}

static void executeRequest(String name, Callback callback) {
    new Thread(() -> {
        try {
            System.out.println(" Starting... " + name);
            Thread.sleep(2000);
            System.out.println(" Finishing... " + name);
            callback.onSuccess();
        } catch (InterruptedException ignored) {}
    }).start();
}

I want to convert this to an RX Completable. To do so I use Completable#create(). The implementation of the CompletableEmitter calls the executeRequest passing of implementation of the Callback that signals when the request has finished.

I also print a log trace when subscribed to help me debugging.

static Completable createRequestCompletable(String name) {
        return Completable.create(e -> executeRequest(name, e::onComplete))
                .doOnSubscribe(d -> System.out.println("Subscribed to " + name));
}

This works as expected. The Completable completes only after the "request" finishes and the callback is invoked.

Problem is that when subscribing to this these completables in a trampoline scheduler, it does not wait for the first request to finish before subscribing to the second request.

This code:

final Completable c1 = createRequestCompletable("1");
c1.subscribeOn(Schedulers.trampoline()).subscribe();

final Completable c2 = createRequestCompletable("2");
c2.subscribeOn(Schedulers.trampoline()).subscribe();

Outputs:

Subscribed to 1
    Starting... 1 
Subscribed to 2
    Starting... 2 
    Finishing... 1 
    Finishing... 2

As you see, it subscribes to the second Completable before the first Completable has completed, even if I'm subscribing in trampoline.

I'd like to queue the completables, so that the second waits for the first to finish, outputting this:

Subscribed to 1
    Starting... 1 
    Finishing... 1
Subscribed to 2
    Starting... 2  
    Finishing... 2

I'm sure the problem is related to the work being done in a worker thread. If the implementation of the Completable does not spawn a new thread it works as expected. But this is legacy code and what I'm trying to do is adapting it to RX without modifying.

NOTE: the requests are executed in different points of the program - I cannot use andThen or concat to implement the serialised execution.

user_789110 · Answer

I have managed to execute the Completables sequentially by explicitly blocking the subscription Thread with a Latch. But I don't think this is the idiomatic way to do it in RX, and I still don't understand why I need to do this and the Thread is not blocked until the Completable completes.

static Completable createRequestCompletable(String name) {
    final CountDownLatch latch = new CountDownLatch(1);
    return Completable.create(e -> {
        executeRequest(name, () -> {
            e.onComplete();
            latch.countDown();
        });
        latch.await();
    })
    .doOnSubscribe(disposable -> System.out.println("Subscribed to " + name));
}

Code adapted to RX Completable not blocking onSubscribe thread

1 Answers