Update STOMP MESSAGE frames with messageId

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

@@ -17,6 +17,7 @@
 package org.springframework.web.messaging;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -85,7 +86,7 @@ public class PubSubHeaders {
 
 		this.messageHeaders = readOnly ? messageHeaders : new HashMap<String, Object>(messageHeaders);
 		this.rawHeaders = this.messageHeaders.containsKey(RAW_HEADERS) ?
-				(Map<String, String>) messageHeaders.get(RAW_HEADERS) : new HashMap<String, String>();
+				(Map<String, String>) messageHeaders.get(RAW_HEADERS) : Collections.<String, String>emptyMap();
 
 		if (this.messageHeaders.get(MESSAGE_TYPE) == null) {
 			this.messageHeaders.put(MESSAGE_TYPE, MessageType.MESSAGE);

spring-websocket/src/main/java/org/springframework/web/messaging/service/AbstractMessageService.java

@@ -75,7 +75,8 @@ public abstract class AbstractMessageService {
 					logger.trace("Processing notification: " + message);
 				}
 
-				MessageType messageType = (MessageType) message.getHeaders().get("messageType");
+				PubSubHeaders headers = new PubSubHeaders(message.getHeaders(), true);
+				MessageType messageType = headers.getMessageType();
 				if (messageType == null || messageType.equals(MessageType.OTHER)) {
 					processOther(message);
 				}

spring-websocket/src/main/java/org/springframework/web/messaging/stomp/StompHeaders.java

@@ -224,6 +224,9 @@ public class StompHeaders extends PubSubHeaders {
 		if (subscriptionId != null) {
 			getRawHeaders().put(SUBSCRIPTION, subscriptionId);
 		}
+		if (StompCommand.MESSAGE.equals(getStompCommand()) && (getMessageId() == null)) {
+			getRawHeaders().put(MESSAGE_ID, getMessageHeaders().get(ID).toString());
+		}
 	}
 
 }