How to cancel Multi upstream

[thebe14]

I started discussions in the past (#967, #711) asking how to cancel the upstream. I never got an answer, I only got a particular solution (cancel in case of error by throwing an exception). But a solution to cancel is needed outside of error cases.

The task

Pick a service instance to handle some task. Which means load from a configuration file the settings for multiple/all service instances, instantiate a REST client for each, then try them one-by-one:

• call each instance and ask it if it can handle the task,
• once I find one that answers affirmatively, call it again to do the work, and
• ignore the other service instances.

My approach

I know that the multi.select().first(Predicate) operator forwards the items while the passed predicate returns true and could be used to only send the first configuration downstream that qualifies for the task (avoiding the need to cancel from the downstream handler). However, my decision is more costly than a predicate (needs at least one network call). So I need to do the query (if a service instance can handle the workload) in the downstream handler.

I use a Multi for a stream of service configs in a Quarkus project (as all service configurations are loaded at once with a class that maps the YAML config file). Thus I will have to decide in the downstream handler when to stop.

Note that this can also be solved by using a generator for the Multi, but that will just postpone the discussion I want to have (and I don't know how to implement it anyway with async calls in the generator - if someone can show me, I'd be grateful).

My sample implementation uses a REST client for Postman GET /echo to simulate the query to the service, asking if it can do the work (see repo https://github.com/thebe14/async-test).

The problem

I do not know how to cancel the upstream from a downstream handler. Help!

public interface IService {
    public abstract String getName();
    public abstract Uni<String> canDoWork();
    public abstract Uni<Boolean> doWork();
}

public class Service {
    private static final Logger LOG = Logger.getLogger(Test.class);
    private static Random random = new Random();
    Postman postman = null;
    private String name;

    public Uni<String> canDoWork() {
        if(null == this.postman) {
            LOG.error("No client");
            return Uni.createFrom().failure(new RuntimeException("client"));
        }

        var result = Uni.createFrom().nullItem()
            .ifNoItem()
                .after(Duration.ofMillis(2000))
                .failWith(new RuntimeException("timeout"))
            .chain(unused -> {
                // Start our dummy query here
                LOG.infof("Start test %s", this.name);
                return this.postman.echo();
            }) // Uni<RequestDetails>
            .chain(info -> {
                // Got result of our dummy query, simulate what we need to return
                var testResult = random.nextBoolean();
                LOG.infof("Finished test %s (%s)", this.name, testResult ? "true" : "false");
                return testResult ? Uni.createFrom().item(this.name) : Uni.createFrom().nullItem();
            }) // Uni<String>
            .onFailure().invoke(e -> {
                LOG.errorf("Error in test %s: " + e, this.name);
            });

        return result;
    }

    public Uni<Boolean> doWork() { ... }
}

@RequestScoped
public class MyResource {
    private static final Logger LOG = Logger.getLogger(MyResource.class);
    private List<IService> services;
    private Subscription subscription;

    @GET
    @Path("/pick-first")
    public Uni<Boolean> pickFirstValid() {
        var result = Multi.createFrom().iterable(this.services)
            .onSubscription().invoke(sub -> {
                // Save the subscription, so we can cancel later
                LOG.info("Subscribed");
                this.subscription = sub;
            })
            .onItem().transformToUniAndConcatenate(service -> {
                // Query service
                LOG.infof("Querying service %s", service.getName());
                return service.canDoWork();
            }) // Multi<String>
            .onItem().transformToUniAndConcatenate(serviceName -> {
                // Query finished
                if(null != serviceName && !serviceName.isBlank()) {
                    LOG.infof("Found service (%s)", serviceName);

                    // Cancel upstream
                    if(null != this.subscription)
                        subscription.cancel();

                    var service = findService(serviceName);
                    return service.doWork();
                }

                return Uni.createFrom().item(false);
            }) // Multi<Boolean>
            .onFailure().invoke(e -> {
                LOG.error("Cannot pick candidate");
            })
            .collect() // MultiCollect<Boolean>
            .in(() -> Boolean.FALSE, (acc, candidate) -> {
                acc = acc || candidate;
            });

        return result;
    }
}

[jponge]

• If you do plain Mutiny then you have to subscribe and pass a subscriber. The subscriber is given a subscription that can request elements and cancel the subscription.
• In Quarkus the framework is doing the subscription on your behalf, so if you want to cancel then you have to get the subscription (see onSubscription()) and eventually call cancel() on it. Also see https://quarkus.io/guides/quarkus-reactive-architecture

[thebe14]

@jponge Any feedback on my approach above? Can I make this work?
I think the documentation is in acute need of an example on how to cancel in the downstream handler.

[jponge]

Have you checked https://javadoc.io/doc/io.smallrye.reactive/mutiny/1.7.0/io/smallrye/mutiny/groups/MultiOnSubscribe.html to get a reference to the subscription you might eventually cancel?

[thebe14]

Yes, I did. This works great, however extra care needs to be taken in case the Quarkus resource implementing the endpoint(s) is not request scoped, as that means pickFirstValid() can be called multiple times on the same MyResource instance, in which case you need to store all subscriptions (e.g. in a Map<String, Subscription> where the key is a UUID you generate on entry into method pickFirstValid) and pick the right one to cancel. Thank you!

I also updated the code above (and the repo) with the suggested solution.

[thebe14]

@jponge I hastily declared that the above solution works. It deadlocks. If I try to cancel the multi stream, it will exit the .onItem().transformToUniAndConcatenate() handler where I call cancel(), but then it is deadlocked. I do not even know how to debug this further. Help!

[thebe14]

Another problem is the accumulator I am using. The one in the example above is not working as expected, the acc is not preserved. I would like to see a suggestion that works (return a single boolean value that is an OR of the result of all processed stream items), without having to write an accumulator class to wrap a boolean.

[thebe14]

My solution to the accumulator issue:

public class BooleanAccumulator {
    private boolean state;

    public BooleanAccumulator() { this.state = false; }
    public void accumulateAny(boolean b) { this.state = this.state || b; }
    public void accumulateAll(boolean b) { this.state = this.state && b; }
    public boolean get() { return this.state; }
}

@RequestScoped
public class MyResource {

    @GET
    @Path("/pick-first")
    public Uni<Boolean> pickFirstValid() {
        var result = Multi.createFrom().iterable(this.services)
            .onSubscription().invoke(sub -> { ... })
            ...
            .collect() // MultiCollect<Boolean>
            .in(BooleanAccumulator::new, (acc, candidate) -> {
                acc.accumulateAny(candidate);
            })
            .onItem().transform(BooleanAccumulator::get);

        return result;
    }
}