From 4886edd10b4df22ed4e61f24e0c11124b255687b Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 20 Mar 2015 17:34:31 -0400 Subject: [PATCH] Send STOMP ERROR if external broker not available Issue: SPR-12820 --- .../stomp/StompBrokerRelayMessageHandler.java | 22 +++++-- .../StompBrokerRelayMessageHandlerTests.java | 57 ++++++++++++++++++- 2 files changed, 71 insertions(+), 8 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java index a9a1f496df..fc74ddc360 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java +++ b/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 java.util.concurrent.atomic.AtomicInteger; 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 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler 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; } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java b/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java index 6a4f2d7b6e..2115303aa7 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java +++ b/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 class StompBrokerRelayMessageHandlerTests { } @Test - public void testOutboundMessage() throws Exception { + public void testOutboundMessageIsEnriched() throws Exception { this.brokerRelay.start(); @@ -170,6 +170,59 @@ public class StompBrokerRelayMessageHandlerTests { 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 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 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 getVoidFuture() { ListenableFutureTask futureTask = new ListenableFutureTask<>(new Callable() {