Tests waiting mechanisms in reactive event-loop context

[silviu-negoita]

Hello,

I have a test in quarkus 3.0, where I need to interact with the DB and websockets.

I am using @TestReactiveTransaction and UniAsserter asserter which are needed for hibernate reactive.
Here are a list of operations I have for a test:

1. Mock certain services. E.g. auth
2. Insert something in db -> needs event-loop thread and I am using asserter.execute()
3. Connect ws -> this does some data processing on server on @OnOpen method.
4. I need to wait for step 3 to be done to test my asserts. e.g. my sessions are activated in the database, user is authorized for that ws session, etc.

Between 3 and 4 I am losing the reactivity chain (I cannot write my code in reactive manner, chaining the operations). I have to manually wait for some time for the data to get processed.

Before quarkus 3, I had two mechanisms to wait:

• awaitility:org.awaitility to wait my data to be processed after websocket connects. e.g. await().atMost(1, SECONDS).until(() -> securedSessions.size() == 1)
• LinkedBlockingDeque<String> and it's poll method: serverMessages.poll(4000, MILLISECONDS)

But starting with quarkus 3, I cannot do this anymore because it blocks event-loops threads.

Are there any suggestions how I can go further with this? Maybe if I can switch back and forth from a event-loop thread to a regular thread where I can block?

What I tried:

1. Using ManagedExecutor to switch to an worker thread, but then I cannot chain the operations (those from step 2).
2. Get rid of @TestReactiveTransaction and UniAsserter asserter and try to manually activate vertx context (on event-loop thread) when db interactions are needed. But again, I cannot chain the operations.

Ref I used:
https://quarkus.io/guides/hibernate-reactive-panache#testing
https://quarkus.io/guides/websockets#websocket-clients

[silviu-negoita]

Update:

I think I came with a solution:

The initial thread is event-loop because I used @RunOnVertxContext

var chain = Uni.createFrom().nullItem()
                 // switch to executor thread
                .emitOn(Infrastructure.getDefaultWorkerPool())
                .onItem().invoke(() -> {
                  // my awaits
                }).emitOn(command -> { 
                  // switch back to event loop thread
                    Context context = Vertx.currentContext();
                    context.runOnContext(x -> command.run());
                });
asserter.execute(() -> chain)
             .assertThat(() -> 
                        Panache.withTransaction(() -> { // persist using panache reactive)
                , (res) -> System.out.println("DONE"));

I am open to new ideas, because I find this approach closer to a workaround rather to an official solution.

Later update:

More on that, I created this to wrap blocking operations:

    Uni<Void> runBlocking(Runnable runnable) {
        return Uni.createFrom().nullItem()
                .emitOn(Infrastructure.getDefaultWorkerPool())
                .onItem().invoke(runnable).emitOn(command -> {
                    Context context = Vertx.currentContext();
                    context.runOnContext(x -> command.run());
                }).replaceWithVoid();
    }

Now it can be used like this:

asserter.execute(() -> runBlocking(() -> // blocking))
             .assertThat(() -> Entity.persist(), (res) -> asserts)
             .execute(() -> runBlocking(() -> // other blocking))
)