Simple broker sends notice after disconnect

Before this change the simple broker simply removed subscriptions
upon receiving a DISCONNECT message assuming it was a result of
a client STOMP WebSocket session ending.

However, if the server-side application sends a DISCONNECT to
the broker in order to terminate a session, the STOMP WebSocket
session could remain unware without any further action. This
change ensures the simple broker sends a DISCONNECT_ACK message
downstream whenever it receives a DISCONNECT.

Issue: SPR-12288

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageType.java

@@ -39,6 +39,8 @@ public enum SimpMessageType {
 
 	DISCONNECT,
 
+	DISCONNECT_ACK,
+
 	OTHER;
 
 }

spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java

@@ -169,6 +169,11 @@ else if (SimpMessageType.DISCONNECT.equals(messageType)) {
 				logger.debug("Processing " + accessor.getShortLogMessage(EMPTY_PAYLOAD));
 			}
 			this.subscriptionRegistry.unregisterAllSubscriptions(sessionId);
+			SimpMessageHeaderAccessor disconnectAck = SimpMessageHeaderAccessor.create(SimpMessageType.DISCONNECT_ACK);
+			initHeaders(disconnectAck);
+			disconnectAck.setSessionId(sessionId);
+			Message<byte[]> messageOut = MessageBuilder.createMessage(EMPTY_PAYLOAD, disconnectAck.getMessageHeaders());
+			getClientOutboundChannel().send(messageOut);
 		}
 		else if (SimpMessageType.SUBSCRIBE.equals(messageType)) {
 			if (logger.isDebugEnabled()) {

spring-messaging/src/test/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandlerTests.java

@@ -16,6 +16,10 @@
 
 package org.springframework.messaging.simp.broker;
 
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
 import java.util.Collections;
 
 import org.junit.Before;
@@ -24,17 +28,13 @@
 import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
-
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessageType;
 import org.springframework.messaging.support.MessageBuilder;
 
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
-
 /**
  * Unit tests for SimpleBrokerMessageHandler.
  *
@@ -115,7 +115,12 @@ public void subcribeDisconnectPublish() {
 		this.messageHandler.handleMessage(createMessage("/foo", "message1"));
 		this.messageHandler.handleMessage(createMessage("/bar", "message2"));
 
-		verify(this.clientOutboundChannel, times(3)).send(this.messageCaptor.capture());
+		verify(this.clientOutboundChannel, times(4)).send(this.messageCaptor.capture());
+
+		Message<?> captured = this.messageCaptor.getAllValues().get(0);
+		assertEquals(SimpMessageType.DISCONNECT_ACK, SimpMessageHeaderAccessor.getMessageType(captured.getHeaders()));
+		assertEquals(sess1, SimpMessageHeaderAccessor.getSessionId(captured.getHeaders()));
+
 		assertCapturedMessage(sess2, "sub1", "/foo");
 		assertCapturedMessage(sess2, "sub2", "/foo");
 		assertCapturedMessage(sess2, "sub3", "/bar");

spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java

@@ -393,6 +393,10 @@ else if (accessor instanceof SimpMessageHeaderAccessor) {
 			if (SimpMessageType.CONNECT_ACK.equals(stompAccessor.getMessageType())) {
 				stompAccessor = convertConnectAcktoStompConnected(stompAccessor);
 			}
+			else if (SimpMessageType.DISCONNECT_ACK.equals(stompAccessor.getMessageType())) {
+				stompAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
+				stompAccessor.setMessage("Session closed.");
+			}
 			else if (stompAccessor.getCommand() == null || StompCommand.SEND.equals(stompAccessor.getCommand())) {
 				stompAccessor.updateStompCommandAsServerMessage();
 			}