Thread and context problem with Hibernate Reactive Panache, RESTeasy Reactive and WebSockets.

[SetoKaiba]

I have LiveManager.ensureRoom method called indirectly by LiveServerEndpoint and directly by LiveResource.
The call from LiveServerEndpoint wil be always ok.
The call from LiveResource will be broken after the call from LiveServerEndpoint.

#32336
And also, this is another problem about Hibernate Reactive Panache and WebSockets as well.
The call from onMessage will not have a current vertx context.

It seems that reactive is said to be a best practise of Quarkus while a lot of extensions become reactive.
But it seems that the collaboration among extensions with reactive is with a lot of problem.
They will work separately. But it become problem after mix them together.

Can any one help on this, please?

2023-04-03 13:56:16,841 ERROR [io.qua.ver.htt.run.QuarkusErrorHandler] (vert.x-eventloop-thread-3) HTTP Request to /live/streams failed, error id: 8facd388-fd2c-4e28-8b09-040e0cb9f656-1: java.lang.IllegalStateException: HR000069: Detected use of the reactive Session from a different Thread than the one which was used to open the reactive Session - this suggests an invalid integration; original thread [218]: 'vert.x-eventloop-thread-3' current Thread [252]: 'executor-thread-2'

LiveManager.java

    public static void onOpen(Long roomId, String userId, String name, Session session, Logger logger) {
        Panache.withSession(() -> ensureRoom(roomId, userId, name, session, logger)).subscribe().with(liveUser -> {
            loggedIn(roomId, session, liveUser, logger);
        });
    }

    public static Uni<LiveUser> ensureRoom(Long roomId, String userId, String name, Session session, Logger logger) {
        if (ROOM_MAP.containsKey(roomId)) {
            LiveRoom liveRoom = ROOM_MAP.get(roomId);
            return ensureUser(userId, name, session, liveRoom, logger);
        } else {
            return Room.<Room>findById(roomId).onItem().transformToUni(room -> {
                if (room == null) {
                    return Uni.createFrom().nullItem();
                } else {
                    LiveRoom liveRoom = new LiveRoom(logger, room);
                    ROOM_MAP.put(roomId, liveRoom);
                    return ensureUser(userId, name, session, liveRoom, logger);
                }
            });
        }
    }

LiveServerEndpoint.java

    @OnOpen
    public void onOpen(Session session, @PathParam("roomId") Long roomId) {
        sharedLogger.getLogger().infof("onOpen: %s, %s", session, roomId);
        String id, name;
        if (securityIdentity.isAnonymous()) {
            id = UUID.randomUUID().toString();
            name = id;
        } else {
            id = jsonWebToken.getSubject();
            name = jsonWebToken.getName();
        }
        LiveManager.onOpen(roomId, id, name, session, sharedLogger.getLogger());
    }

LiveResource.java

    @Path("/streams")
    @POST
    @WithSession
    public Uni<LiveResponse> streams(JsonObject jsonObject) {
        String app = jsonObject.getString("app");
        String stream = jsonObject.getString("stream");
        logger.info(jsonObject);
        if (app.equals("live") && stream.equals("livestream")) {
            return Uni.createFrom().item(new LiveResponse(LiveResponse.CODE_OK, "ok"));
        } else {

            String action = jsonObject.getString("action");
            boolean publishing = action.equals("on_publish");
            String param = jsonObject.getString("param");
            param = param.replaceAll("\\?", "");
            HashMap<String, String> paramMap = new HashMap<>();
            for (String s : param.split("&")) {
                String[] child = s.split("=");
                paramMap.put(child[0], child[1]);
            }
            String streamCode = paramMap.get("streamCode");
            try {
                Long roomId = Long.parseLong(app);
                return User.<User>findById(UUID.fromString(stream)).onItem().transformToUni(user -> {
                    if (user.streamCode.toString().equals(streamCode)) {
                        return LiveManager.ensureRoom(roomId, stream, user.name, null, sharedLogger.getLogger())
                                .onItem().transformToUni(liveUser -> onPublish(publishing, liveUser));
                    } else {
                        return Uni.createFrom().item(new LiveResponse(LiveResponse.CODE_STREAM_KEY_CHECK_FAILS, "code stream key check fails"));
                    }
                });
            } catch (NumberFormatException numberFormatException) {
                return Uni.createFrom().item(new LiveResponse(LiveResponse.CODE_ROOM_ID_FORMAT_ERROR, "room id format error"));
            }
        }
    }

[quarkus-bot[bot]]

/cc @DavideD (hibernate-reactive), @FroMage (panache,resteasy-reactive), @Sanne (hibernate-reactive), @Sgitario (resteasy-reactive), @gavinking (hibernate-reactive), @geoand (resteasy-reactive), @loicmathieu (panache), @stuartwdouglas (resteasy-reactive)

[SetoKaiba]

I found that the problem is caused by thread group changes. It sometimes use the executor-thread, sometimes use the vert.x-eventloop-thread. It's said that in RESTEasy Reactive: Everything runs on the event loop unless you add @Blocking to the endpoint. I have a OIDC authorizaiton settings. Does it matter? Why is it sometimes in the executor thread, sometimes in the eventloop thread?

And I found this post here mentioned the authorization may affact this.
https://groups.google.com/g/quarkus-dev/c/GnccGHS8Jyo
Maybe this is related as well.
#31606

[SetoKaiba]

After a deep dive, I found tha it's caused by my code Uni.createFrom().future(xxx). The xxx is a returned future of session.getAsyncRemote().sendBinary(yyy). The session is in executor thread. How can I unify the future and uni together to make it reactive?

[SetoKaiba]

Use emitOn to switch back to the original thread solve the problem.

        Context context = Vertx.currentContext();
        return Uni.createFrom().future(session.getAsyncRemote().sendBinary(byteBuf.nioBuffer()))
                .invoke(() -> logger.infof("%s sent %s", id, session))
                .emitOn(command -> context.runOnContext(x -> command.run()));