Send STOMP ERROR if external broker not available

Issue: SPR-12820

Author: rstoyanchev

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -26,6 +26,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessageType;
@@ -399,12 +400,21 @@ protected void handleMessageInternal(Message<?> message) {
 				throw new MessageDeliveryException("Message broker not active. Consider subscribing to " +
 						"receive BrokerAvailabilityEvent's from an ApplicationListener Spring bean.");
 			}
-			SimpMessageType messageType = SimpMessageHeaderAccessor.getMessageType(message.getHeaders());
-			if (logger.isErrorEnabled() && SimpMessageType.CONNECT.equals(messageType)) {
-				logger.error("Broker not active. Ignoring " + message);
+			StompConnectionHandler handler = this.connectionHandlers.get(sessionId);
+			if (handler != null) {
+				handler.sendStompErrorFrameToClient("Broker not available.");
+				handler.clearConnection();
 			}
-			else if (logger.isDebugEnabled()) {
-				logger.debug("Broker not active. Ignoring " + message);
+			else {
+				StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.ERROR);
+				if (getHeaderInitializer() != null) {
+					getHeaderInitializer().initHeaders(accessor);
+				}
+				accessor.setSessionId(sessionId);
+				accessor.setUser(SimpMessageHeaderAccessor.getUser(message.getHeaders()));
+				accessor.setMessage("Broker not available.");
+				MessageHeaders headers = accessor.getMessageHeaders();
+				getClientOutboundChannel().send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers));
 			}
 			return;
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -146,7 +146,7 @@ public void testDestinationExcluded() throws Exception {
 	}
 
 	@Test
-	public void testOutboundMessage() throws Exception {
+	public void testOutboundMessageIsEnriched() throws Exception {
 
 		this.brokerRelay.start();
 
@@ -170,6 +170,59 @@ public void testOutboundMessage() throws Exception {
 		assertEquals("joe", actualHeaders.getUser().getName());
 	}
 
+	// SPR-12820
+
+	@Test
+	public void testConnectWhenBrokerNotAvailable() throws Exception {
+
+		this.brokerRelay.start();
+		this.brokerRelay.stopInternal();
+
+		String sessionId = "sess1";
+		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
+		headers.setSessionId(sessionId);
+		headers.setUser(new TestPrincipal("joe"));
+		this.brokerRelay.handleMessage(MessageBuilder.createMessage(new byte[0], headers.getMessageHeaders()));
+
+		Message<byte[]> actual = this.outboundChannel.getMessages().get(0);
+		StompHeaderAccessor actualHeaders = StompHeaderAccessor.getAccessor(actual, StompHeaderAccessor.class);
+		assertEquals(StompCommand.ERROR, actualHeaders.getCommand());
+		assertEquals(sessionId, actualHeaders.getSessionId());
+		assertEquals("joe", actualHeaders.getUser().getName());
+		assertEquals("Broker not available.", actualHeaders.getMessage());
+	}
+
+	@Test
+	public void testSendAfterBrokerUnavailable() throws Exception {
+
+		this.brokerRelay.start();
+
+		String sessionId = "sess1";
+		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
+		headers.setSessionId(sessionId);
+		headers.setUser(new TestPrincipal("joe"));
+		this.brokerRelay.handleMessage(MessageBuilder.createMessage(new byte[0], headers.getMessageHeaders()));
+
+		assertEquals(2, this.brokerRelay.getConnectionCount());
+
+		this.brokerRelay.stopInternal();
+
+		headers = StompHeaderAccessor.create(StompCommand.SEND);
+		headers.setSessionId(sessionId);
+		headers.setUser(new TestPrincipal("joe"));
+		headers.setDestination("/foo");
+		this.brokerRelay.handleMessage(MessageBuilder.createMessage(new byte[0], headers.getMessageHeaders()));
+
+		assertEquals(1, this.brokerRelay.getConnectionCount());
+
+		Message<byte[]> actual = this.outboundChannel.getMessages().get(0);
+		StompHeaderAccessor actualHeaders = StompHeaderAccessor.getAccessor(actual, StompHeaderAccessor.class);
+		assertEquals(StompCommand.ERROR, actualHeaders.getCommand());
+		assertEquals(sessionId, actualHeaders.getSessionId());
+		assertEquals("joe", actualHeaders.getUser().getName());
+		assertEquals("Broker not available.", actualHeaders.getMessage());
+	}
+
 
 	private static ListenableFutureTask<Void> getVoidFuture() {
 		ListenableFutureTask<Void> futureTask = new ListenableFutureTask<>(new Callable<Void>() {