Add support for multi node architecture usging RabbitMQ as a full STOMP broker

Author: jorgeacetozi

ebook-chat/pom.xml

@@ -68,9 +68,9 @@
 			<version>1.1.2</version>
 		</dependency>
 		<dependency>
-			<groupId>org.webjars</groupId>
+			<groupId>org.webjars.bower</groupId>
 			<artifactId>stomp-websocket</artifactId>
-			<version>2.3.3</version>
+			<version>2.3.4</version>
 		</dependency>
 		<dependency>
 			<groupId>org.webjars</groupId>
@@ -104,9 +104,27 @@
 			<artifactId>spring-security-test</artifactId>
 		</dependency>
 		<dependency>
-		    <groupId>org.webjars</groupId>
-		    <artifactId>noty</artifactId>
-		    <version>2.2.4</version>
+			<groupId>org.webjars</groupId>
+			<artifactId>noty</artifactId>
+			<version>2.2.4</version>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-net</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-all</artifactId>
+			<version>4.1.7.Final</version>
+		</dependency>
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-transport-native-epoll</artifactId>
+			<version>4.1.7.Final</version>
 		</dependency>
 	</dependencies>
 

ebook-chat/src/main/java/br/com/jorgeacetozi/ebookChat/chatroom/api/ChatRoomController.java

@@ -49,20 +49,20 @@ public ModelAndView join(@PathVariable String chatRoomId, Principal principal) {
 		return modelAndView;
 	}
 
-	@SubscribeMapping("/connected-users")
+	@SubscribeMapping("/connected.users")
 	public List<ChatRoomUser> listChatRoomConnectedUsersOnSubscribe(SimpMessageHeaderAccessor headerAccessor) {
 		String chatRoomId = headerAccessor.getSessionAttributes().get("chatRoomId").toString();
 		return chatRoomService.findById(chatRoomId).getConnectedUsers();
 	}
 
-	@SubscribeMapping("/old-messages")
+	@SubscribeMapping("/old.messages")
 	public List<InstantMessage> listOldMessagesFromUserOnSubscribe(Principal principal,
 			SimpMessageHeaderAccessor headerAccessor) {
 		String chatRoomId = headerAccessor.getSessionAttributes().get("chatRoomId").toString();
 		return instantMessageService.findAllInstantMessagesFor(principal.getName(), chatRoomId);
 	}
 
-	@MessageMapping("/send-message")
+	@MessageMapping("/send.message")
 	public void sendMessage(@Payload InstantMessage instantMessage, Principal principal,
 			SimpMessageHeaderAccessor headerAccessor) {
 		String chatRoomId = headerAccessor.getSessionAttributes().get("chatRoomId").toString();

ebook-chat/src/main/java/br/com/jorgeacetozi/ebookChat/configuration/WebSocketConfigSpringSession.java

@@ -1,5 +1,6 @@
 package br.com.jorgeacetozi.ebookChat.configuration;
 
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.simp.config.MessageBrokerRegistry;
 import org.springframework.scheduling.annotation.EnableScheduling;
@@ -13,12 +14,23 @@
 @EnableWebSocketMessageBroker
 public class WebSocketConfigSpringSession extends AbstractSessionWebSocketMessageBrokerConfigurer<ExpiringSession> {
 
+	@Value("${ebook.chat.relay.host}")
+	private String relayHost;
+
+	@Value("${ebook.chat.relay.port}")
+	private Integer relayPort;
+
 	protected void configureStompEndpoints(StompEndpointRegistry registry) {
 		registry.addEndpoint("/ws").withSockJS();
 	}
 
 	public void configureMessageBroker(MessageBrokerRegistry registry) {
-		registry.enableSimpleBroker("/queue/", "/topic/");
+		registry.enableStompBrokerRelay("/queue/", "/topic/")
+			.setUserDestinationBroadcast("/topic/unresolved.user.dest")
+			.setUserRegistryBroadcast("/topic/registry.broadcast")
+			.setRelayHost(relayHost)
+			.setRelayPort(relayPort);
+
 		registry.setApplicationDestinationPrefixes("/chatroom");
 	}
 }

ebook-chat/src/main/java/br/com/jorgeacetozi/ebookChat/utils/Destinations.java

@@ -5,15 +5,15 @@ public class Destinations {
 	public static class ChatRoom {
 
 		public static String publicMessages(String chatRoomId) {
-			return "/topic/" + chatRoomId + "/public-messages";
+			return "/topic/" + chatRoomId + ".public.messages";
 		}
 
 		public static String privateMessages(String chatRoomId) {
-			return "/queue/" + chatRoomId + "/private-messages";
+			return "/queue/" + chatRoomId + ".private.messages";
 		}
 
 		public static String connectedUsers(String chatRoomId) {
-			return "/topic/" + chatRoomId + "/connected-users";
+			return "/topic/" + chatRoomId + ".connected.users";
 		}
 	}
 }

ebook-chat/src/main/resources/application.yml

@@ -41,6 +41,12 @@ flyway.enabled: true
 security.headers.cache: false 
 server.session.timeout: 600
 
+ebook:
+  chat:
+    relay:
+      host: localhost
+      port: 61613
+
 ---
 
 spring:
@@ -69,4 +75,10 @@ spring:
       repositories:
         enabled: true
 
-flyway.enabled: true
+flyway.enabled: true
+
+ebook:
+  chat:
+    relay:
+      host: localhost
+      port: 61613

ebook-chat/src/main/resources/templates/chatroom.html

@@ -90,7 +90,7 @@ <h1 class="panel-title" th:text="#{chatroom.users}">Users</h1>
 <script src="/webjars/jquery/3.1.1/jquery.min.js"></script>
 <script src="/webjars/bootstrap/3.3.7/js/bootstrap.min.js"></script>
 <script src="/webjars/sockjs-client/1.1.2/sockjs.min.js"></script>
-<script src="/webjars/stomp-websocket/2.3.3/stomp.min.js"></script>
+<script src="/webjars/stomp-websocket/2.3.4/lib/stomp.min.js"></script>
 <script src="/webjars/noty/2.2.4/jquery.noty.packaged.min.js"></script>
 
 <script th:inline="javascript">
@@ -116,12 +116,12 @@ <h1 class="panel-title" th:text="#{chatroom.users}">Users</h1>
 		enableInputMessage();
 		successMessage("Your WebSocket connection was successfuly established!")
 
-		stompClient.subscribe('/chatroom/connected-users', updateConnectedUsers);
-		stompClient.subscribe('/chatroom/old-messages', oldMessages);
+		stompClient.subscribe('/chatroom/connected.users', updateConnectedUsers);
+		stompClient.subscribe('/chatroom/old.messages', oldMessages);
 
-		stompClient.subscribe('/topic/' + chatRoomId + '/public-messages', publicMessages);
-		stompClient.subscribe('/user/queue/' + chatRoomId + '/private-messages', privateMessages);
-		stompClient.subscribe('/topic/' + chatRoomId + '/connected-users', updateConnectedUsers);
+		stompClient.subscribe('/topic/' + chatRoomId + '.public.messages', publicMessages);
+		stompClient.subscribe('/user/queue/' + chatRoomId + '.private.messages', privateMessages);
+		stompClient.subscribe('/topic/' + chatRoomId + '.connected.users', updateConnectedUsers);
 	}
 
 	function stompFailure(error) {
@@ -218,7 +218,7 @@ <h1 class="panel-title" th:text="#{chatroom.users}">Users</h1>
 				'toUser' : spanSendTo.text()
 			}
 		}
-		stompClient.send("/chatroom/send-message", {}, JSON.stringify(instantMessage));
+		stompClient.send("/chatroom/send.message", {}, JSON.stringify(instantMessage));
 		inputMessage.val("").focus();
 	}
 

ebook-chat/src/test/java/br/com/jorgeacetozi/ebookChat/integrationTests/test/AbstractIntegrationTest.java

@@ -5,20 +5,26 @@
 import org.testcontainers.containers.GenericContainer;
 
 public class AbstractIntegrationTest {
-	
+
 	@ClassRule
 	public static final GenericContainer cassandra = new FixedHostPortGenericContainer("cassandra:3.0")
 			.withFixedExposedPort(9042, 9042)
 			.withNetworkMode("host");
-	
+
 	@ClassRule
 	public static final GenericContainer mysql = new FixedHostPortGenericContainer("mysql:5.7")
 			.withFixedExposedPort(3306, 3306)
 			.withEnv("MYSQL_DATABASE", "ebook_chat")
-			.withEnv("MYSQL_ROOT_PASSWORD","root")
+			.withEnv("MYSQL_ROOT_PASSWORD", "root")
 			.withNetworkMode("host");
-	
+
 	@ClassRule
 	public static final GenericContainer redis = new FixedHostPortGenericContainer("redis:3.0.6")
-			.withFixedExposedPort(6379, 6379);
+			.withFixedExposedPort(6379, 6379)
+			.withNetworkMode("host");
+
+	@ClassRule
+	public static final GenericContainer rabbitmq = new FixedHostPortGenericContainer("jorgeacetozi/rabbitmq-stomp:3.6")
+			.withFixedExposedPort(61613, 61613)
+			.withNetworkMode("host");
 }

ebook-chat/src/test/java/br/com/jorgeacetozi/ebookChat/unitTests/utils/DestinationsTest.java

@@ -13,16 +13,16 @@ public class DestinationsTest {
 
 	@Test
 	public void shouldGetPublicMessagesDestination() {
-		assertThat(Destinations.ChatRoom.publicMessages("123"), is("/topic/" + chatRoomId + "/public-messages"));
+		assertThat(Destinations.ChatRoom.publicMessages("123"), is("/topic/" + chatRoomId + ".public.messages"));
 	}
 
 	@Test
 	public void shouldGetPrivateMessagesDestination() {
-		assertThat(Destinations.ChatRoom.privateMessages("123"), is("/queue/" + chatRoomId + "/private-messages"));
+		assertThat(Destinations.ChatRoom.privateMessages("123"), is("/queue/" + chatRoomId + ".private.messages"));
 	}
 
 	@Test
 	public void shouldGetConnectedUsersDestination() {
-		assertThat(Destinations.ChatRoom.connectedUsers("123"), is("/topic/" + chatRoomId + "/connected-users"));
+		assertThat(Destinations.ChatRoom.connectedUsers("123"), is("/topic/" + chatRoomId + ".connected.users"));
 	}
 }