Support STOMP DISCONNECT with receipt

Issue: SPR-11599
This commit is contained in:
Rossen Stoyanchev
2014-06-28 23:09:14 -04:00
parent 7dc2b2927e
commit 7da3fb4ce6
2 changed files with 43 additions and 6 deletions

View File

@@ -759,7 +759,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override @Override
public void onSuccess(Void result) { public void onSuccess(Void result) {
if (accessor.getCommand() == StompCommand.DISCONNECT) { if (accessor.getCommand() == StompCommand.DISCONNECT) {
clearConnection(); afterDisconnectSent(accessor);
} }
} }
@Override @Override
@@ -775,6 +775,21 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
return future; return future;
} }
/**
* After a DISCONNECT there should be no more client frames so we can
* close the connection pro-actively. However, if the DISCONNECT has a
* receipt header we leave the connection open and expect the server will
* respond with a RECEIPT and then close the connection.
*
* @see <a href="http://stomp.github.io/stomp-specification-1.2.html#DISCONNECT">
* STOMP Specification 1.2 DISCONNECT</a>
*/
private void afterDisconnectSent(StompHeaderAccessor accessor) {
if (accessor.getReceipt() == null) {
clearConnection();
}
}
/** /**
* Clean up state associated with the connection and close it. * Clean up state associated with the connection and close it.
* Any exception arising from closing the connection are propagated. * Any exception arising from closing the connection are propagated.

View File

@@ -223,11 +223,6 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
startActiveMqBroker(); startActiveMqBroker();
this.eventPublisher.expectBrokerAvailabilityEvent(true); this.eventPublisher.expectBrokerAvailabilityEvent(true);
// TODO The event publisher assertions show that the broker's back up and the system relay session
// has reconnected. We need to decide what we want the reconnect behaviour to be for client relay
// sessions and add further message sending and assertions as appropriate. At the moment any client
// sessions will be closed and an ERROR from will be sent.
} }
@Test @Test
@@ -249,6 +244,21 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
assertTrue("Unexpected messages: " + this.responseHandler.queue, this.responseHandler.queue.isEmpty()); assertTrue("Unexpected messages: " + this.responseHandler.queue, this.responseHandler.queue.isEmpty());
} }
@Test
public void disconnectWithReceipt() throws Exception {
logger.debug("Starting test disconnectWithReceipt()");
MessageExchange connect = MessageExchangeBuilder.connect("sess1").build();
this.relay.handleMessage(connect.message);
this.responseHandler.expectMessages(connect);
MessageExchange disconnect = MessageExchangeBuilder.disconnectWithReceipt("sess1", "r123").build();
this.relay.handleMessage(disconnect.message);
this.responseHandler.expectMessages(disconnect);
}
private static class TestEventPublisher implements ApplicationEventPublisher { private static class TestEventPublisher implements ApplicationEventPublisher {
@@ -403,6 +413,18 @@ public class StompBrokerRelayMessageHandlerIntegrationTests {
return new MessageExchangeBuilder(message); return new MessageExchangeBuilder(message);
} }
public static MessageExchangeBuilder disconnectWithReceipt(String sessionId, String receiptId) {
StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.DISCONNECT);
headers.setSessionId(sessionId);
headers.setReceipt(receiptId);
Message<?> message = MessageBuilder.createMessage(new byte[0], headers.getMessageHeaders());
MessageExchangeBuilder builder = new MessageExchangeBuilder(message);
builder.expected.add(new StompReceiptFrameMessageMatcher(sessionId, receiptId));
return builder;
}
public MessageExchangeBuilder andExpectMessage(String sessionId, String subscriptionId) { public MessageExchangeBuilder andExpectMessage(String sessionId, String subscriptionId) {
Assert.isTrue(SimpMessageType.MESSAGE.equals(headers.getMessageType())); Assert.isTrue(SimpMessageType.MESSAGE.equals(headers.getMessageType()));
String destination = this.headers.getDestination(); String destination = this.headers.getDestination();