Fix blocking calls#8
Conversation
RawSanj
left a comment
RawSanj
left a comment
There was a problem hiding this comment.
Can you please work on the suggested changes?
PS: Sorry for the late reply and thanks for the PR 👍🏼
| 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.
| .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.
c4f6135 to
371c6a9
Compare
2 participants
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.