Reactor-core: What if the inner method of operators is blocking? May drag all process down.

[ZhangSanFengByGit]

Code Example

@Override
public void onNext(T t) {
    if (done) {
        Operators.onNextDropped(t, actual.currentContext());
        return;
    }

    R v;

    try {
        v = Objects.requireNonNull(mapper.apply(t),
                "The mapper returned a null value.");
    }
    catch (Throwable e) {
        Throwable e_ = Operators.onNextError(t, e, actual.currentContext(), s);
        if (e_ != null) {
            onError(e_);
        }
        else {
            s.request(1);
        }
        return;
    }

    actual.onNext(v);
}

Question

Above is the code of onNext method of FluxMap, which is just for an example.
Here, My question is that the whole subscribe->onSubscribe->request->onNext stream mechanism seems to be synchronize to me. So if the inner mapper method here, which is set by programmer, is in blocking style, for instance, contains I/O blocking, the whole publisher-operator-subscriber process may be blocked since they are all in one thread.

So is that a drawback of reactor that the inner method of operators must be non-blocking?

[smaldini]

Good catch, it is also documented in the reference guide and in general its a requirement for "Reactive Streams" libraries such as rxjava and even in other non-blocking alternatives like Kotlin Coroutines.

In Reactor (or RxJava), you can protect such blocking callback by immediately preceding a signal with a thread boundary:

• publishOn for onNext, onComplete, onError, request
• subscribeOn for subscribe, request, onSubscribe

[smaldini]

Check out https://github.com/reactor/BlockHound to detect such calls - it is an increasingly popular tool used by a growing community !