INT-4360: Add ClientStompEncoder support

The `StompSubProtocolHandler` explicitly sets `stompCommand` header
to the `MESSAGE` value ignoring any client inputs.
In this case the message is treated as from the server and ignored on
the STOMP Broker side from the client session.

* Introduce `ClientStompEncoder` for the client side to be injected
into the `StompSubProtocolHandler` for the proper client side messages
encoding/decoding.
Override `stompCommand` header to the `SEND` value if it is `MESSAGE`
before encoding to the `byte[]` to send to the session

JIRA: https://jira.spring.io/browse/INT-4360

**Cherry-pick to 4.3.x**

Fix WebSocket test to rely on the proper client config class
and don't pick up the server config unconditionally in the test context

spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/ClientStompEncoder.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.websocket.support;
+
+import java.util.Map;
+
+import org.springframework.messaging.simp.stomp.StompCommand;
+import org.springframework.messaging.simp.stomp.StompEncoder;
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+
+/**
+ * A {@link StompEncoder} extension to prepare a message for sending/receiving
+ * before/after encoding/decoding when used from WebSockets client side.
+ * For example it updates the {@code stompCommand} header from the {@code MESSAGE}
+ * to {@code SEND} frame, which is the case of
+ * {@link org.springframework.web.socket.messaging.StompSubProtocolHandler}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.3.13
+ */
+public class ClientStompEncoder extends StompEncoder {
+
+	@Override
+	public byte[] encode(Map<String, Object> headers, byte[] payload) {
+		if (StompCommand.MESSAGE.equals(headers.get("stompCommand"))) {
+			StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.SEND);
+			stompHeaderAccessor.copyHeadersIfAbsent(headers);
+			headers = stompHeaderAccessor.getMessageHeaders();
+		}
+		return super.encode(headers, payload);
+	}
+
+}

spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java

@@ -55,7 +55,6 @@ import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
-import org.springframework.integration.test.support.LogAdjustingTestSupport;
 import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
 import org.springframework.integration.websocket.ClientWebSocketContainer;
 import org.springframework.integration.websocket.IntegrationWebSocketContainer;
@@ -63,11 +62,13 @@ import org.springframework.integration.websocket.TomcatWebSocketTestServer;
 import org.springframework.integration.websocket.event.ReceiptEvent;
 import org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter;
 import org.springframework.integration.websocket.outbound.WebSocketOutboundMessageHandler;
+import org.springframework.integration.websocket.support.ClientStompEncoder;
 import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.messaging.handler.annotation.SendTo;
@@ -83,7 +84,7 @@ import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Controller;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.socket.client.WebSocketClient;
 import org.springframework.web.socket.client.standard.StandardWebSocketClient;
@@ -106,10 +107,10 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport;
  *
  * @since 4.1
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
-@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
-public class StompIntegrationTests extends LogAdjustingTestSupport {
+@ContextConfiguration(classes = StompIntegrationTests.ClientConfig.class)
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class StompIntegrationTests {
 
 	@Value("#{server.serverContext}")
 	private ApplicationContext serverContext;
@@ -129,11 +130,6 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 	@Qualifier("webSocketEvents")
 	private QueueChannel webSocketEvents;
 
-	public StompIntegrationTests() {
-		super("org.springframework", "org.springframework.integration", "org.apache.catalina");
-	}
-
-
 	@Test
 	public void sendMessageToController() throws Exception {
 		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
@@ -156,6 +152,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 
 		SimpleController controller = this.serverContext.getBean(SimpleController.class);
 		assertTrue(controller.latch.await(20, TimeUnit.SECONDS));
+		assertEquals(StompCommand.SEND.name(), controller.stompCommand);
 	}
 
 	@Test
@@ -331,7 +328,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 
 	@Configuration
 	@EnableIntegration
-	public static class ContextConfiguration {
+	public static class ClientConfig {
 
 		@Bean
 		public TomcatWebSocketTestServer server() {
@@ -350,7 +347,9 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 
 		@Bean
 		public SubProtocolHandler stompSubProtocolHandler() {
-			return new StompSubProtocolHandler();
+			StompSubProtocolHandler stompSubProtocolHandler = new StompSubProtocolHandler();
+			stompSubProtocolHandler.setEncoder(new ClientStompEncoder());
+			return stompSubProtocolHandler;
 		}
 
 		@Bean
@@ -396,10 +395,11 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 
 	// WebSocket Server part
 
-	@Target({ElementType.TYPE})
+	@Target({ ElementType.TYPE })
 	@Retention(RetentionPolicy.RUNTIME)
 	@Controller
 	private @interface IntegrationTestController {
+
 	}
 
 	@IntegrationTestController
@@ -407,8 +407,11 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 
 		private final CountDownLatch latch = new CountDownLatch(1);
 
+		private String stompCommand;
+
 		@MessageMapping("/simple")
-		public void handle() {
+		public void handle(@Header("stompCommand") String stompCommand) {
+			this.stompCommand = stompCommand;
 			this.latch.countDown();
 		}
 

spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/WebSocketClientTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -54,7 +54,7 @@ import org.springframework.messaging.support.GenericMessage;
 import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.web.socket.client.WebSocketClient;
 import org.springframework.web.socket.client.standard.StandardWebSocketClient;
 import org.springframework.web.socket.messaging.StompSubProtocolHandler;
@@ -65,10 +65,11 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport;
 
 /**
  * @author Artem Bilan
+ *
  * @since 4.1
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = WebSocketClientTests.ClientConfig.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class WebSocketClientTests {
 
@@ -96,7 +97,7 @@ public class WebSocketClientTests {
 
 	@Configuration
 	@EnableIntegration
-	public static class ContextConfiguration {
+	public static class ClientConfig {
 
 		@Bean
 		public TomcatWebSocketTestServer server() {

spring-integration-websocket/src/test/java/org/springframework/integration/websocket/server/WebSocketServerTests.java

@@ -75,7 +75,7 @@ import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.WebSocketMessage;
@@ -101,8 +101,8 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport;
  *
  * @since 4.1
  */
-@ContextConfiguration(classes = WebSocketServerTests.ContextConfiguration.class)
-@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = WebSocketServerTests.ClientConfig.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class WebSocketServerTests {
 
@@ -186,7 +186,7 @@ public class WebSocketServerTests {
 
 	@Configuration
 	@EnableIntegration
-	public static class ContextConfiguration {
+	public static class ClientConfig {
 
 		@Bean
 		public TomcatWebSocketTestServer server() {

src/reference/asciidoc/web-sockets.adoc

@@ -433,3 +433,12 @@ Defaults to `false`.
 
 
 <13> See the same option on the `<int-websocket:outbound-channel-adapter>`.
+
+[[client-stomp-encoder]]
+=== ClientStompEncoder
+
+Starting with _version 4.3.13_, the `ClientStompEncoder` is provided as an extension of standard `StompEncoder` for using on client side of the WebSocket Channel Adapters.
+An instance of the `ClientStompEncoder` must be injected into the `StompSubProtocolHandler` for proper client side message preparation.
+One of the problem of the default `StompSubProtocolHandler` that it was designed for the server side, so it updates the `SEND` `stompCommand` header into `MESSAGE` as it must be by the STOMP protocol from server side.
+If client doesn't send its messages in the proper `SEND` web socket frame, some STOMP brokers won't accept them.
+The purpose of the `ClientStompEncoder`, in this case, is to override `stompCommand` header to the `SEND` value before encoding the message to the `byte[]`.