Cannot cancel Multi item stream

[thebe14]

Context

I am using a Multi for an event stream in a Quarkus project. Processing each item is complicated (takes a lot of time, calls several APIs in different services) and may fail. In that case, I would like to abort processing the rest of the items in the stream. No matter what I do, I cannot seem to abort on a failed item.

Description

My approach:

    public class Test {
        private static final Logger LOG = Logger.getLogger(Test.class);

        public void multiTest() {
            List<String> items = Arrays.asList("one", "two", "three", "four", "five", "six", "seven", "eight", "nine");
            Multi<String> stream = Multi.createFrom().iterable(items);

            Cancellable cancel = stream
                .onItem().transform(item -> {
                    if (item.equals("two")) {
                        //cancel.cancel();
                        return Multi.createFrom().failure(new Exception(item));
                    }
                    return item;
                })
                .onCompletion().invoke(() -> {
                    LOG.info("Completed");
                })
                .onFailure().invoke(() -> {
                    LOG.info("Failed");
                })
                .onCancellation().invoke(() -> {
                    LOG.info("Cancelled");
                })
                .subscribe().with(item -> {
                    // Trace items
                    if (item.getClass().equals(String.class)) {
                        String s = (String) item;
                        LOG.infof("Handled %s", s);
                    } else
                        LOG.infof("Stop item in stream");
                });
        }
    }

The output is always this:

Handled one
Stop item in stream
Handled three
Handled four
Handled five
Handled six
Handled seven
Handled eight
Handled nine
Completed

Q1: Shouldn't this stop after the second item in the stream? How can I achieve that?
Q2: Are items in the stream processed sequentially or are processed all in parallel?

After we can identify why this is not working, including an example on how to stop processing items in a stream in the documentation would be essential.

Additional details

I tried using both Exception and RuntimeException in the "stop" item, no cigar.

The idea to process items in the transform callback, so that I have a chance to replace an item with a "stop" item, came from this issue. The documentation is not helpful, I started reading closed issues on GitHub out of desperation.

The commented cancel.cancel(); cannot be called from the transform handler, as it warns that cancel might not have been initialized yet. So I figure that this Cancellable can only be used from outside of the handlers to stop the stream (issue a cancellation event). The idea for this came from this documentation page, however that should clarify that the Cancellable cannot be used in event handlers.

[cescoffier]

I'm wondering how your example is compiling.
Instead of return Multi.createFrom().failure(new Exception(item));, throws a RuntimeException.

[thebe14]

Thanks, this works. I explained above how I came to this attempt. Basically because the documentation is not helpful, so I guessed. My initial try was to throw an Exception, I saw that that does not compile. I had no idea that throwing a RuntimeException would work.

[cescoffier]

Yes, the doc does not cover cancellation.
Basically, in your case, you want to fail, which will cancel the upstream and send the failure downstream.

[thebe14]

Can I throw a RuntimeException from any handler to cancel (the up)stream?