Changes needed to broadcast only to room

[kinjalnyc]

This is really awesome, thank you.

Quick question, if I want to modify this to support multiple rooms how should I proceed ?

[RawSanj]

Hi @kinjalnyc - sorry for the late reply.

I didn't spend much time, but here's one way you can create multiple rooms, and have chat messages delivered to the targeted ones.

1. Update HanlderMapping to accept query in url:

        @Bean
	public HandlerMapping webSocketHandlerMapping(ChatWebSocketHandler webSocketHandler) {
		Map<String, WebSocketHandler> map = new HashMap<>();
		map.put("/redis-chat/**", webSocketHandler);
		SimpleUrlHandlerMapping handlerMapping = new SimpleUrlHandlerMapping();
		handlerMapping.setCorsConfigurations(Collections.singletonMap("*", new CorsConfiguration().applyPermitDefaultValues()));
		handlerMapping.setOrder(1);
		handlerMapping.setUrlMap(map);
		return handlerMapping;
	}

2. Add room field in the message:

      public class ChatMessage {
	      private Integer id;
	      private String message;
	      private String hostname;
	      private Long usersOnline;
	      private String room;
      }

3. Update RedisChatMessagePublisher.publishChatMessage() method to accept room:

         public Mono<Long> publishChatMessage(String message, String room) {
		Integer totalChatMessage = chatMessageCounter.incrementAndGet();
		return Mono.fromCallable(() -> {
			try {
				return InetAddress.getLocalHost().getHostName();
			} catch (UnknownHostException e) {
				log.error("Error getting hostname.", e);
			}
			return "localhost";
		}).map(hostName -> {
			ChatMessage chatMessage = new ChatMessage(totalChatMessage, message, hostName, activeUserCounter.get(), room);
			String chatString = "EMPTY_MESSAGE";
			try {
				chatString = objectMapper.writeValueAsString(chatMessage);
			} catch (JsonProcessingException e) {
				log.error("Error converting ChatMessage {} into string", chatMessage, e);
			}
			return chatString;
		}).flatMap(chatString -> {
			// Publish Message to Redis Channels
			return reactiveStringRedisTemplate.convertAndSend(MESSAGE_TOPIC, chatString)
				.doOnSuccess(aLong -> log.debug("Total of {} Messages published to Redis Topic.", totalChatMessage))
				.doOnError(throwable -> log.error("Error publishing message.", throwable));
		});
	}

5. Update ChatWebSocketHandler.handle() method to get room from the URL:
   public Mono handle(WebSocketSession webSocketSession) {

    String room = webSocketSession.getHandshakeInfo().getUri().getQuery().replace("room=", "");
    
    Flux<WebSocketMessage> sendMessageFlux = chatMessageFluxSink.filter(chatMessage -> chatMessage.getRoom().equals(room)).flatMap(objectStringConverter::objectToString)
    	.map(webSocketSession::textMessage)
    	.doOnError(throwable -> log.error("Error Occurred while sending message to WebSocket.", throwable));
    Mono<Void> outputMessage = webSocketSession.send(sendMessageFlux);
   
    Mono<Void> inputMessage = webSocketSession.receive()
    	.flatMap(webSocketMessage -> redisChatMessagePublisher.publishChatMessage(webSocketMessage.getPayloadAsText(), room))
    	.doOnSubscribe(subscription -> {
    		long activeUserCount = activeUserCounter.incrementAndGet();
    		log.info("User '{}' Connected. Total Active Users: {}", webSocketSession.getId(), activeUserCount);
    		chatMessageSink.tryEmitNext(new ChatMessage(0, "CONNECTED", "CONNECTED", activeUserCount, room));
    	})
    	.doOnError(throwable -> log.error("Error Occurred while sending message to Redis.", throwable))
    	.doFinally(signalType -> {
    		long activeUserCount = activeUserCounter.decrementAndGet();
    		log.info("User '{}' Disconnected. Total Active Users: {}", webSocketSession.getId(), activeUserCount);
    		chatMessageSink.tryEmitNext(new ChatMessage(0, "DISCONNECTED", "DISCONNECTED", activeUserCount, room));
    	})
    	.then();
   
    return Mono.zip(inputMessage, outputMessage).then();
   

   }

6. And finally update the UI forms, to accept room before sending messages. You can run below JS in browser console to send and receive messages:

let host = location.hostname + (location.port ? ':' + location.port : '');
let wsProtocol = location.protocol === "https:" ? "wss://" : "ws://";

// connecting room=one
let room1 = new WebSocket(wsProtocol + host + "/redis-chat?room=one");
room1.onopen = openEvent => {
	console.log("room1 opened", openEvent);
};
room1.onmessage = messageEvent => {
	console.log("Message: from room1 ", messageEvent);
};

// connecting room=two
let room2 = new WebSocket(wsProtocol + host + "/redis-chat?room=two");
room2.onopen = openEvent => {
	console.log("room2 opened", openEvent);
};
room2.onmessage = messageEvent => {
	console.log("Message: from room2 ", messageEvent);
};

// Send messages:
room1.send('This message goes to all connected to room-1');
room2.send('This message goes to all connected to room-2');
room1.send('This message goes to all connected to room-1');

Hope this help!