Reactor-core: Custom Publisher and potential flatMap bug

[ikajinic]

Hello,

I've come across a potential bug with custom Publisher and flatMap operation where any operation after flatMap is not executed. I've created failing test case which showcases the issue. Example might be a bit contrived and I'm not sure I'm using the framework "as intended", so appologies for that. Here is the test case:

public class PotentialFlatMapBug {
    @Test(expected = RuntimeException.class)
    public void thatExceptionIsThrown() {
        Receiver receiver = new Receiver();
        new Handler(receiver);

        for (int i = 0; i < 10; i++) {
            receiver.receive(i);
        }
    }
    public static class Handler {
        private final Receiver receiver;
        public Handler(Receiver receiver) {
            this.receiver = receiver;
            initHandler();
        }
        public void initHandler() {
            Flux.from(receiver)
                .flatMap(Mono::just)
                .log()
                .map(this::throwException)
                .subscribe();
        }
        public Mono<Void> throwException(Integer i) {
            throw new RuntimeException("Exception");
        }
    }
    public static class Receiver implements Publisher<Integer> {
        private Subscriber<? super Integer> subscriber;
        public void receive(Integer input) {
            subscriber.onNext(input);
        }
        @Override
        public void subscribe(Subscriber<? super Integer> s) {
            this.subscriber = s;
        }
    }
}

Test will fail having expected a RuntimeException to show up. Also there will be nothing logged in console, no other exceptions, no onNext() calls or anything.

[simonbasle]

Rule number 1 of the Reactor club: don't write your own Publisher. Even though the interface is simple, the set of rules about interactions between all these reactive streams interface is not.

Here for example, there are several issues:

1. your publisher never completes (might be intentional, but I don't think so)
2. your publisher short-circuits the subscription process. it should call onSubscribe on the Subscriber and pass it a Subscription the subscriber can use to request data or cancel
3. without the subscription, or if the publisher ignores the subscription's request() calls, the publisher ignores backpressure.
4. your test expected is too broad, it could be accepting any kind of exception. better to narrow it inside a try { ...; fail(); } catch { ... } block.

Even though it only fixes item 2) above, just modifying the subscribe method to call s.onSubscribe demonstrates that something will start happening:

@Override
public void subscribe(Subscriber<? super Integer> s) {
    this.subscriber = s;
    s.onSubscribe(Operators.emptySubscription());
}

Removing the expected from the unit test, however, we see that our exception was indeed thrown:

10:42:12.143 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
10:42:12.158 [main] INFO  reactor.Flux.FlatMap.1 - onSubscribe(FluxFlatMap.FlatMapMain)
10:42:12.160 [main] INFO  reactor.Flux.FlatMap.1 - request(unbounded)
10:42:12.171 [main] INFO  reactor.Flux.FlatMap.1 - onNext(0)
10:42:12.172 [main] INFO  reactor.Flux.FlatMap.1 - cancel()

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.RuntimeException: Exception

Caused by: java.lang.RuntimeException: Exception
    at PotentialFlatMapBug$Handler.throwException(PotentialFlatMapBug.java:36)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:165)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.emitScalar(FluxFlatMap.java:380)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:349)
    at PotentialFlatMapBug$Receiver.receive(PotentialFlatMapBug.java:42)
    at PotentialFlatMapBug.thatExceptionIsThrown(PotentialFlatMapBug.java:19)

The ErrorCallbackNotImplemented comes from the fact that in Handler your subscribe doesn't define what to do with each item and in case of errors.

In Conclusion: Do not roll your own Publisher

I'd recommend that you look at Flux.create() (if you want to bridge an existing API into a Flux) or various Processors (if you just want to feed values to something manually and have these values represented as a Flux).

[ikajinic]

Hi Simon,

Thank you for a detailed explanation. Calling onSubscribe indead helped.

As far as your points:

1. I did infact want a publisher that never completes. I have an existing app that receives constant influx of data via UDP and I've used reactor to implement a scatter-gather pattern.

2. throwException was a bit silly here, but it was the easiest thing to implement and show the actual bug. In real app this would be some transformation or passing along data to some outside system. Point was that this method was never called.

With all that said, would you still consider this a bug since it just sort of failed silently without any exceptions or log entries?

[simonbasle]

No, this is not a bug, this is a badly implemented Publisher. As I said, strive to use either Flux.create or a Processor for this use case (I think a DirectProcessor would do the trick). There will be a lot of other problems with that kind of implementation, and the use of the emptySubscription() I did was just for demonstration purposes, not a proper fix.

[ikajinic]

Will do. Thank you.

[simonbasle]

No problem, writing proper Publisher and operators is hard 😄