Fix blocking calls #8
Conversation
| long activeUserCount = activeUserCounter.decrementAndGet(); | ||
| log.info("User '{}' Disconnected. Total Active Users: {}", webSocketSession.getId(), activeUserCount); | ||
| chatMessageSink.tryEmitNext(new ChatMessage(0, "DISCONNECTED", "DISCONNECTED", activeUserCount)); | ||
| }).subscribeOn(Schedulers.boundedElastic()); |
There was a problem hiding this comment.
You are missing .subscribe() here. The code inside Mono.fromRunnable won't execute unless explicitly subscribed or blocked.
| @@ -23,7 +24,9 @@ public RouterFunction<ServerResponse> htmlRouter(@Value("classpath:/static/index | |||
| return route(GET("/"), request -> ok().contentType(MediaType.TEXT_HTML).bodyValue(html)) | |||
| .andRoute(POST("/message"), request -> request.bodyToMono(Message.class) | |||
| .flatMap(message -> redisChatMessagePublisher.publishChatMessage(message.getMessage())) | |||
| .flatMap(aLong -> ServerResponse.ok().bodyValue(new Message("Message Sent Successfully!.")))); | |||
| .flatMap(aLong -> ServerResponse.ok().bodyValue(new Message("Message Sent Successfully!."))) | |||
| .subscribeOn(Schedulers.boundedElastic())); | |||
There was a problem hiding this comment.
Do we really need to make this run on Schedulers.boundedElastic() Did the BlockHound detected blocking call over here or inside the RedisChatMessagePublisher?
There was a problem hiding this comment.
In the first stack trace (first screenshot in the post), we see that the error propagates to RedisChatMessagePublisher but was triggered by WebHttpHandler, line 26 right? :)
There was a problem hiding this comment.
In the first stack trace (first screenshot in the post), we see that the error propagates to RedisChatMessagePublisher but was triggered by WebHttpHandler, line 26 right? :)
The other way round. The actual blocking call is at line - https://github.com/RawSanj/spring-redis-websocket/blob/master/src/main/java/com/github/rawsanj/messaging/RedisChatMessagePublisher.java#L35C11-L35C28 the Atomic Increment is blocking.
I think we should be good if we move that blocking call Integer totalChatMessage = chatMessageCounter.incrementAndGet() inside the Mono.fromCallable() and then add the .subscribeOn(Schedulers.boundedElastic()) in the end.
Arooba-git
force-pushed
the
fix-blocking-calls
branch
from
c4f6135
to
371c6a9
Compare
Hi! 👋
Apparently the chat module has some blocking calls (as detected by BlockHound):
This PR addresses these blocking code to ensure the pipeline remain reactive.