Split async handling to multiple classes/methods

[thebe14]

Context

From an API endpoint implemented in a Quarkus 2.9.1 project with resteasy-reactive-jackson and rest-client-reactive-jackson, I am calling a REST client that returns Unis. If I keep all the reactive processing in my endpoint, this approach works great. However, if I have to split the reactive processing between multiple methods (implemented by different classes), this becomes difficult. I found an approach that sort of works, but causes errors and Mutiny to drop exceptions. Help!

Everything in one method

This works great (although I welcome any suggestion on how to do it more elegantly). Even when I signal failure throwing a RuntimeException it is handled as expected:

@Path("/myresource")
public class MyResource {
    private static final Logger LOG = Logger.getLogger(MyResource.class);
    private SomeRestClient restClient;

    @GET	
    public Response doTest() {
        try {
            CompletableFuture<Response> response = new CompletableFuture<>();
            Uni<String> start = Uni.createFrom().item("");
            
            start
                .chain(unused-> {
                    // Do test
                    return restClient.someTaskAsync();
                })
                .chain(s -> {
                    // Simulate error
                    if(some_condition)
                        throw new RuntimeException("problem");
                })
                .chain(s -> {
                    // Success
                    response.complete(Response.ok(s).build());
                    return Uni.createFrom().nullItem();
                })
                .onFailure().invoke(e -> {
                    LOG.error("Failed test");
                    if (!response.isDone())
                        response.complete(Response.ok(e).status(Status.EXPECTATION_FAILED).build());
                })
                .subscribe();

            // Wait until test completes (possibly with error)
            Response r = response.get();
            return r;
        } catch (InterruptedException e) {
            return Response.status(Status.INTERNAL_SERVER_ERROR, "interrupted");
        } catch (ExecutionException e) {
            return Response.status(Status.INTERNAL_SERVER_ERROR, "execution");
        }
    }
}

Processing split between methods

When I have to move a part of the async processing to a different method, the only solution I found was to synchronously wait for the processing step to finish, then artificially wrap the result with an Uni. My implementation:

public interface ITest {
    public abstract Uni<String> test();
}

public class MyException extends RuntimeException {}

public class Test implements ITest {
    private static final Logger LOG = Logger.getLogger(Test.class);
    private SomeRestClient restClient;

    public Uni<String> test() {
        if(null == restClient)
            throw new MyException("client");

        AtomicReference<Uni<String>> result = new AtomicReference<>();
        Uni<String> info = this.restClient.someTaskAsync();

        info
            .ifNoItem().after(Duration.ofMillis(100)).failWith(() -> {
                LOG.error("Timeout");
                return new MyException("timeout");
             )}
            .chain(s -> {
                // Success
                result.set(Uni.createFrom().item(s));
                return Uni.createFrom().nullItem();
            })
            .onFailure().invoke(e -> {
                LOG.error(e);
                result.set(Uni.createFrom().failure(e));
            })
            .await().indefinitely();

        return result.get();
    }
}

@Path("/myresource")
public class MyResource {
    private static final Logger LOG = Logger.getLogger(MyResource.class);
    private ITest tester;

    @GET	
    public Response doTest() {
        try {
            CompletableFuture<Response> response = new CompletableFuture<>();
            Uni<String> start = Uni.createFrom().item("");
            
            start
                .chain(s -> {
                    // Do test
                    return tester.test();
                })
                .chain(s -> {
                    // Success
                    response.complete(Response.ok(s).build());
                    return Uni.createFrom().nullItem();
                })
                .onFailure().invoke(e -> {
                    LOG.error("Failed test");
                    if (!response.isDone())
                        response.complete(Response.ok(e).status(Status.EXPECTATION_FAILED).build());
                })
                .subscribe();

            // Wait until test completes (possibly with error)
            Response r = response.get();
            return r;
        } catch (InterruptedException e) {
            return Response.status(Status.INTERNAL_SERVER_ERROR, "interrupted");
        } catch (ExecutionException e) {
            return Response.status(Status.INTERNAL_SERVER_ERROR, "execution");
        }
    }
}

Although this works, in the sense that in Test.test() both throwing an exception directly or failing the reactive processing (e.g. on timeout) results in control correctly reaching the onFailure() handler in MyResource.doTest(), Mutiny drops an exception with this error message:

ERROR [fo.ba.Test]] (executor-thread-1) Timeout
ERROR [fo.ba.Test]] (executor-thread-1) foo.bar.MyException
ERROR [fo.ba.MyResource]] (executor-thread-0) Failed test
ERROR [io.qu.mu.ru.MutinyInfrastructure]] (executor-thread-0) Mutiny had to drop the following exception: foo.bar.MyException

Q1: Why is Mutiny unhappy and how can I get this working clean, without errors?
Q2. Am I really blocking 2 threads while MyResource.doTest() executes?
Q3: Any way to implement Test.test() without synchronously waiting for the operation to finish?

[jponge]

Some elements from a purely Mutiny perspective, but I don't understand everything here.

• Q1 : you throw a MyException from a call operator. This is not the proper way to forward exceptions. call expects a Uni, so you should return a Uni that eventually has a value, or a Uni that has a failure (see Uni.createFrom()).
• Q2 : Yes, you block because you call blocking methods (CompletableFuture.get()).
• Q3 : If you return a Uni then you shouldn't block in your method. Just return a Uni, and when that Uni completes a HTTP response will eventually be made. See https://quarkus.io/guides/getting-started-reactive

[thebe14]

@jponge Thank you very much for the answers. Unfortunately, although your answers are logical, it just muddies my understanding more.

For Q1 I throw an exception at the beginning of the method. But even if I rewrite it to return a failure Uni (see below), Mutiny still drops an exception.

public class Test implements ITest {
    private static final Logger LOG = Logger.getLogger(Test.class);
    private SomeRestClient restClient;

    public Uni<String> test() {
         return Uni.createFrom.failure(new MyException("client"));
    }
}

For Q2 in MyResource.doTest() I wanted to block, so that when the REST API returns it has some result. I was more curious about Test.test() if that blocks and how to write it to avid blocking. Especially if the processing there starts from Uni<A> on which I have .chain() handlers to transform it into Uni<B> that will be returned by the method. Not clear to me how to write the method to return the Uni wrapping the transformed type.

My take (does not compile):

public class Test implements ITest {
    private static final Logger LOG = Logger.getLogger(Test.class);
    private SomeRestClient restClient;

    public Uni<Boolean> test() {
        if(null == restClient) {
            LOG.error("No client");
            return Uni.createFrom().failure(new MyException("client"));
        }

        Uni<String> info = this.restClient.someTaskAsync();

        info
            .ifNoItem().after(Duration.ofMillis(100)).failWith(() -> {
                LOG.error("Timeout");
                return new MyException("timeout");
             )}
            .chain(s -> {
                // Success
                return Uni.createFrom().item(new Boolean(true));
            })
            .onFailure().invoke(e -> {
                LOG.error(e);
            });

        return info; // <- does not compile, different type
    }
}

[jponge]

Q1 Perhaps something else in the call chain drops the exception, but it's not Mutiny itself.

Q2 What's the actual type of info?

[thebe14]

Q1. If you take the code above from under Processing split between methods and replace Test.test() with the one that returns a failure Uni, then calling MyResource.doTest() e.g. from the Swagger-UI of the Quarkus project, will cause Mutiny to drop the MyException exception.

Q2. String, as declared. I start from a (Uni of) String returned by the REST client I call, but want to return a (Uni of) Boolean from the method.

[jponge]

Try onItem().transform(...)

[thebe14]

I am already using onItem().transformToUni() or that is what I thought chain() is.

[thebe14]

For Q2 where I did not know how to write Test.test() so that it does not block. Especially if the processing starts from Uni<A> on which I have .chain() handlers to transform it into Uni<B> that will be returned by the method. My solution is to create an initial Uni of the type the method will return, add .chain() handlers on that (which are just a shorthand of .onItem().transformToUni() handlers), ignore the input of the first handler (the initial dummy item), then do whatever operations and transformations are needed, provided the last transform eventually delivers the same type as declared. Return the initial dummy Uni from the method.

public class Test implements ITest {
    private static final Logger LOG = Logger.getLogger(Test.class);
    private SomeRestClient restClient;

    public Uni<Boolean> test() {
        if(null == restClient) {
            LOG.error("No client");
            return Uni.createFrom().failure(new MyException("client"));
        }

        Uni<Boolean> result = Uni.createFrom().nullItem(); // <- This gets returned, handlers are not called

        result
            .ifNoItem().after(Duration.ofMillis(100)).failWith(() -> {
                LOG.error("Timeout");
                return new MyException("timeout");
             )}
            .chain(unused -> {
                // Ignore initial dummy placeholder, start the actual processing here
                LOG.info("Start processing");
                Uni<String> info = this.restClient.someTaskAsync();
                return info;
            })
            .chain(s -> {
                // Got result of processing, turn it into what we have to return
                LOG.info("Finished processing");
                return Uni.createFrom().item(new Boolean(s.equals("evrika")));
            })
            .onFailure().invoke(e -> {
                LOG.error(e);
            });

        return result; 
    }
}

However, when I call this deeper async processing from MyResource.doTest(), the behavior is not what I expected. The call I am doing from this modified resource class:

@Path("/myresource")
public class MyResource {
    private static final Logger LOG = Logger.getLogger(MyResource.class);
    private ITest tester;

    @GET	
    public Response doTest() {
        AtomicReference<Response> result = new AtomicReference<>();
        Uni<String> start = Uni.createFrom().item("");            
        start
            .chain(s -> {
                // Do test
                LOG.info("Start test");
                return tester.test(); // <- Returns immediately, does not cause invocation of handlers in Test.test()
            })
            .chain(b -> { // <- We get null passed to this handler
                // Success
                LOG.info("Finished test");
                result.set(Response.ok(b).build());
                return Uni.createFrom().nullItem();
            })
            .onFailure().invoke(e -> {
                LOG.error("Failed test");
                result.set(Response.ok(e).status(Status.EXPECTATION_FAILED).build());
            })
            .await().indefinitely();

        return result;
    }
}

I am expecting this output:

Start test
Start processing
Finished processing
Finished test

Instead I am getting this output:

Start test
Finished test

The handlers in Test.test() are ignored, the initial item created for the Uni in Test.test() is returned.

@jponge Can you please shed a light on what I am doing wrong? I find this a huge problem with the documentation, such an example is badly needed.

[thebe14]

Empirical finding. If I write it like this, handlers in the deep method Test.test() are called:

public class Test implements ITest {
    ...
    public Uni<Boolean> test() {
        ...
        Uni<Boolean> result = Uni.createFrom().nullItem()

            .ifNoItem().after(Duration.ofMillis(100)).failWith(() -> {
                ...
             )}
            .chain(unused -> {
                ...
            })
            .onFailure().invoke(e -> {
                ...
            });

        return result; 
    }
}

[thebe14]

An explanation would be welcome about why it works like this and it does not work as my previous attempt.

[jponge]

Do you have any minimal reproducer project? It’s hard to guess what ITest is, etc.

…

On Mon, Jun 6, 2022, at 21:41, Levente Farkas wrote: An explanation would be welcome about why it works like this and it does not work as my previous attempt. — Reply to this email directly, view it on GitHub <#933 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAAGK2LDXH6TNGBTSR7AGXLVNZH43ANCNFSM5W4YP37Q>. You are receiving this because you were mentioned.Message ID: ***@***.***>

[jponge]

Thanks!

So looking at the code here are a few observations.

• Not sure why you have ITest and Test, it might be simpler to consolidate into 1 single class. But I may be missing some requirement from your larger code base that justifies an abstraction.
• The 2 methods in MyResource shall return Uni<Response> rather than Response. By doing so RestEasy Reactive will subscribe to the provided Uni, which triggers the actual work. Actually, you need to be aware that no computation/processing is ever started until a Uni/Multi is subscribed. RestEasy Reactive does it for you, else you need to do it manually (see .subscribe().with(...).
• Calling .await().indefinitely() does trigger a subscription. However, since you block in the method, you are not leveraging RestEasy Reactive at all.
• chain processes items, not failures. So if a failure is propagated then chain will never see it (but methods from the onFailure group will).

So overall my main feedback is that you are not using RestEasy Reactive and then the rest of the pipelines you create are probably more complicated than they should be.

I'd also suggest looking at the many snippets in https://github.com/jponge/mutiny-workshop-examples, they'll show you some common Mutiny patterns in snippets that should hopefully be self-explanatory.

[jponge]

Complementary feedback: I'd start by commenting out the code, refactoring by doing proper RestEasy Reactive and returning Uni<Response>, and then build processing from there.

[thebe14]

Thanks for the feedback. But you did not answer why the Test.test2() works and Test.test1() does not. My understanding is they are identical, minor syntactical differences that should not matter at runtime.

The reason for the abstraction via an interface (and the need to split the processing between two methods) is that I have to decide at runtime which implementation of the interface will do the actual processing. Moreover, there is some async processing in the endpoint implementation too, before I call through the interface. I think the need to split processing across multiple methods is huge in the real world, the Mutiny docs and guides urgently need to address this (also how to propagate errors/failures in this case).

I will switch the endpoints to return Uni<Response>, or RestEasy reactive as you call it, but I wanted to deal with one thing at a time.

[jponge]

Now I see what's wrong 😃

All methods (chain, etc) return a new Uni (see the javadocs).

So in test2 you return the full pipeline, while in test1 you just return Uni.createFrom().nullItem(). This is expected and documented.

I will switch the endpoints to return Uni, or RestEasy reactive as you call it, but I wanted to deal with one thing at a time.

Well RestEasy Reactive is the web API that you are using, and it's important to understand how it works with Mutiny, see https://quarkus.io/guides/resteasy-reactive#execution-model-blocking-non-blocking

[thebe14]

I can confirm that after converting the endpoints in the resource MyResource to return Uni<Response> no errors are dropped when I signal errors from the deeper async methods, e.g. Test.test2(). Thanks a lot @jponge !

[thebe14]

Final version that works great:

public class Test implements ITest {
    private static final Logger LOG = Logger.getLogger(Test.class);
    private SomeRestClient restClient;

    public Uni<Boolean> test() {
        if(null == restClient) {
            LOG.error("No client");
            return Uni.createFrom().failure(new MyException("client"));
        }

        Uni<Boolean> result = Uni.createFrom().nullItem()
            .ifNoItem().after(Duration.ofMillis(100)).failWith(() -> {
                LOG.error("Timeout");
                return new MyException("timeout");
             )}
            .chain(unused -> {
                // Ignore initial dummy placeholder, start the actual processing here
                LOG.info("Start processing");
                Uni<String> info = this.restClient.someTaskAsync();
                return info;
            })
            .chain(s -> {
                // Got result of processing, turn it into what we have to return
                LOG.info("Finished processing");
                return Uni.createFrom().item(new Boolean(s.equals("evrika")));
            })
            .onFailure().invoke(e -> {
                LOG.error(e);
            });

        return result; 
    }
}

And the Quarkus endpoint that calls it:

@Path("/myresource")
public class MyResource {
    private static final Logger LOG = Logger.getLogger(MyResource.class);
    private ITest tester; // Needs to be set up

    @GET	
    public Uni<Response> doTest() {
        Uni<Response> result = Uni.createFrom().nullItem()
            .chain(unused -> {
                // Do test
                LOG.info("Start test");
                return tester.test();
            })
            .chain(b -> {
                // Success
                LOG.info("Finished test");
                return Uni.createFrom().item(Response.ok(b).build()));
            })
            .onFailure().recoverWithItem(e -> {
                LOG.error("Failed test");
                return Response.ok(e).status(Status.EXPECTATION_FAILED).build();
            });

        return result;
    }
}