Add STOMP broker relay unit tests

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java

@@ -141,17 +141,6 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
 		return this.relayPort;
 	}
 
-	/**
-	 * Configure the TCP client to for managing STOMP over TCP connections to the message
-	 * broker. This is an optional property that can be used to replace the default
-	 * implementation used for example for testing purposes.
-	 * <p>
-	 * By default an instance of {@link ReactorNettyTcpClient} is used.
-	 */
-	public void setTcpClient(TcpOperations<byte[]> tcpClient) {
-		this.tcpClient = tcpClient;
-	}
-
 	/**
 	 * Set the interval, in milliseconds, at which the "system" connection will, in the
 	 * absence of any other data being sent, send a heartbeat to the STOMP broker. A value
@@ -250,11 +239,20 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
 		return this.virtualHost;
 	}
 
+	/**
+	 * Used for unit testing.
+	 */
+	void setTcpClient(TcpOperations<byte[]> tcpClient) {
+		this.tcpClient = tcpClient;
+	}
+
 
 	@Override
 	protected void startInternal() {
 
-		this.tcpClient = new ReactorNettyTcpClient<byte[]>(this.relayHost, this.relayPort, new StompCodec());
+		if (this.tcpClient == null) {
+			this.tcpClient = new ReactorNettyTcpClient<byte[]>(this.relayHost, this.relayPort, new StompCodec());
+		}
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("Initializing \"system\" TCP connection");

spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerIntegrationTests.java

@@ -130,11 +130,6 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
 		logger.debug("Broker stopped");
 	}
 
-
-	// When TCP client is behind interface and configurable:
-	// test "host" header (virtualHost property)
-	// test "/user/.." destination is excluded
-
 	@Test
 	public void publishSubscribe() throws Exception {
 

spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.messaging.simp.stomp;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.messaging.*;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.messaging.simp.SimpMessageType;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.messaging.support.channel.ExecutorSubscribableChannel;
+import org.springframework.messaging.support.tcp.ReconnectStrategy;
+import org.springframework.messaging.support.tcp.TcpConnection;
+import org.springframework.messaging.support.tcp.TcpConnectionHandler;
+import org.springframework.messaging.support.tcp.TcpOperations;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureAdapter;
+import org.springframework.util.concurrent.ListenableFutureTask;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit tests for StompBrokerRelayMessageHandler.
+ *
+ * @author Rossen Stoyanchev
+ */
+public class StompBrokerRelayMessageHandlerTests {
+
+	private StompBrokerRelayMessageHandler brokerRelay;
+
+	private StubTcpOperations tcpClient;
+
+	private StubMessageChannel responseChannel;
+
+
+	@Before
+	public void setup() {
+
+		this.responseChannel = new StubMessageChannel();
+		this.tcpClient = new StubTcpOperations();
+
+		this.brokerRelay = new StompBrokerRelayMessageHandler(this.responseChannel, Arrays.asList("/topic"));
+		this.brokerRelay.setTcpClient(tcpClient);
+	}
+
+
+	@Test
+	public void testVirtualHostHeader() {
+
+		String virtualHost = "ABC";
+		String sessionId = "sess1";
+
+		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
+		headers.setSessionId(sessionId);
+
+		this.brokerRelay.setVirtualHost(virtualHost);
+		this.brokerRelay.start();
+		this.brokerRelay.handleMessage(MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build());
+
+		List<Message<byte[]>> sent = this.tcpClient.connection.messages;
+		assertEquals(2, sent.size());
+
+		StompHeaderAccessor headers1 = StompHeaderAccessor.wrap(sent.get(0));
+		assertEquals(virtualHost, headers1.getHost());
+
+		StompHeaderAccessor headers2 = StompHeaderAccessor.wrap(sent.get(1));
+		assertEquals(sessionId, headers2.getSessionId());
+		assertEquals(virtualHost, headers2.getHost());
+	}
+
+	@Test
+	public void testDestinationExcluded() {
+
+		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
+		headers.setSessionId("sess1");
+		headers.setDestination("/user/daisy/foo");
+
+		this.brokerRelay.start();
+		this.brokerRelay.handleMessage(MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build());
+
+		List<Message<byte[]>> sent = this.tcpClient.connection.messages;
+		assertEquals(1, sent.size());
+		assertEquals(StompCommand.CONNECT, StompHeaderAccessor.wrap(sent.get(0)).getCommand());
+	}
+
+
+	private static ListenableFutureTask<Void> getFuture() {
+		ListenableFutureTask<Void> futureTask = new ListenableFutureTask<>(new Callable<Void>() {
+			@Override
+			public Void call() throws Exception {
+				return null;
+			}
+		});
+		futureTask.run();
+		return futureTask;
+	}
+
+
+	private static class StubTcpOperations implements TcpOperations<byte[]> {
+
+		private StubTcpConnection connection = new StubTcpConnection();
+
+
+		@Override
+		public ListenableFuture<Void> connect(TcpConnectionHandler<byte[]> connectionHandler) {
+			connectionHandler.afterConnected(this.connection);
+			return getFuture();
+		}
+
+		@Override
+		public ListenableFuture<Void> connect(TcpConnectionHandler<byte[]> connectionHandler, ReconnectStrategy reconnectStrategy) {
+			connectionHandler.afterConnected(this.connection);
+			return getFuture();
+		}
+
+		@Override
+		public ListenableFuture<Void> shutdown() {
+			return getFuture();
+		}
+	}
+
+
+	private static class StubTcpConnection implements TcpConnection<byte[]> {
+
+		private final List<Message<byte[]>> messages = new ArrayList<>();
+
+
+		@Override
+		public ListenableFuture<Void> send(Message<byte[]> message) {
+			this.messages.add(message);
+			return getFuture();
+		}
+
+		@Override
+		public void onReadInactivity(Runnable runnable, long duration) {
+		}
+
+		@Override
+		public void onWriteInactivity(Runnable runnable, long duration) {
+		}
+
+		@Override
+		public void close() {
+		}
+	}
+
+
+	private static class StubMessageChannel implements MessageChannel {
+
+		private final List<Message<byte[]>> messages = new ArrayList<>();
+
+
+		@Override
+		public boolean send(Message<?> message) {
+			this.messages.add((Message<byte[]>) message);
+			return true;
+		}
+
+		@Override
+		public boolean send(Message<?> message, long timeout) {
+			this.messages.add((Message<byte[]>) message);
+			return true;
+		}
+	}
+
+}