AMQ-6699 Fix STOMP over WS not encoding header values

When sending STOMP frames out over WS the marshal isn't doing a proper
encode based on the STOMP version in use and so header values can be
transmitted without proper escaping.

activemq-http/src/main/java/org/apache/activemq/transport/ws/StompWSConnection.java

@@ -23,6 +23,7 @@ import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.transport.stomp.StompFrame;
+import org.apache.activemq.transport.stomp.StompWireFormat;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.WebSocketAdapter;
 import org.eclipse.jetty.websocket.api.WebSocketListener;
@@ -40,6 +41,7 @@ public class StompWSConnection extends WebSocketAdapter implements WebSocketList
     private final CountDownLatch connectLatch = new CountDownLatch(1);
 
     private final BlockingQueue<String> prefetch = new LinkedBlockingDeque<String>();
+    private final StompWireFormat wireFormat = new StompWireFormat();
 
     private int closeCode = -1;
     private String closeMessage;
@@ -68,7 +70,7 @@ public class StompWSConnection extends WebSocketAdapter implements WebSocketList
 
     public synchronized void sendFrame(StompFrame frame) throws Exception {
         checkConnected();
-        connection.getRemote().sendString(frame.format());
+        connection.getRemote().sendString(wireFormat.marshalToString(frame));
     }
 
     public synchronized void keepAlive() throws Exception {

activemq-http/src/main/java/org/apache/activemq/transport/ws/jetty9/StompSocket.java

@@ -47,7 +47,7 @@ public class StompSocket extends AbstractStompSocket implements WebSocketListene
     public void sendToStomp(StompFrame command) throws IOException {
         try {
             //timeout after a period of time so we don't wait forever and hold the protocol lock
-            session.getRemote().sendStringByFuture(command.format()).get(getDefaultSendTimeOut(), TimeUnit.SECONDS);
+            session.getRemote().sendStringByFuture(getWireFormat().marshalToString(command)).get(getDefaultSendTimeOut(), TimeUnit.SECONDS);
         } catch (Exception e) {
             throw IOExceptionSupport.create(e);
         }

activemq-http/src/test/java/org/apache/activemq/transport/ws/StompWSTransportTest.java

@@ -301,4 +301,36 @@ public class StompWSTransportTest extends WSTransportTestSupport {
             LOG.info("Caught exception on write of disconnect", ex);
         }
     }
+
+    @Test(timeout = 60000)
+    public void testEscapedHeaders() throws Exception {
+        String connectFrame = "STOMP\n" +
+                              "login:system\n" +
+                              "passcode:manager\n" +
+                              "accept-version:1.1\n" +
+                              "heart-beat:0,0\n" +
+                              "host:localhost\n" +
+                              "\n" + Stomp.NULL;
+
+        wsStompConnection.sendRawFrame(connectFrame);
+        String incoming = wsStompConnection.receive(30, TimeUnit.SECONDS);
+        assertTrue(incoming.startsWith("CONNECTED"));
+
+        String message = "SEND\n" + "destination:/queue/" + getTestName() + "\nescaped-header:one\\ntwo\\cthree\n\n" + "Hello World" + Stomp.NULL;
+        wsStompConnection.sendRawFrame(message);
+
+        String frame = "SUBSCRIBE\n" + "destination:/queue/" + getTestName() + "\n" +
+                       "id:12345\n" + "ack:auto\n\n" + Stomp.NULL;
+        wsStompConnection.sendRawFrame(frame);
+
+        incoming = wsStompConnection.receive(30, TimeUnit.SECONDS);
+        assertTrue(incoming.startsWith("MESSAGE"));
+        assertTrue(incoming.indexOf("escaped-header:one\\ntwo\\cthree") >= 0);
+
+        try {
+            wsStompConnection.sendFrame(new StompFrame(Stomp.Commands.DISCONNECT));
+        } catch (Exception ex) {
+            LOG.info("Caught exception on write of disconnect", ex);
+        }
+    }
 }

activemq-stomp/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java

@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
@@ -74,16 +74,7 @@ public class StompWireFormat implements WireFormat {
         return unmarshal(dis);
     }
 
-    @Override
-    public void marshal(Object command, DataOutput os) throws IOException {
-        StompFrame stomp = (org.apache.activemq.transport.stomp.StompFrame)command;
-
-        if (stomp.getAction().equals(Stomp.Commands.KEEPALIVE)) {
-            os.write(Stomp.BREAK);
-            return;
-        }
-
-        StringBuilder buffer = new StringBuilder();
+    private StringBuilder marshalHeaders(StompFrame stomp, StringBuilder buffer) throws IOException {
         buffer.append(stomp.getAction());
         buffer.append(Stomp.NEWLINE);
 
@@ -95,19 +86,48 @@ public class StompWireFormat implements WireFormat {
             buffer.append(Stomp.NEWLINE);
         }
 
-        // Add a newline to seperate the headers from the content.
+        // Add a newline to separate the headers from the content.
         buffer.append(Stomp.NEWLINE);
 
-        os.write(buffer.toString().getBytes("UTF-8"));
+        return buffer;
+    }
+
+    @Override
+    public void marshal(Object command, DataOutput os) throws IOException {
+        StompFrame stomp = (org.apache.activemq.transport.stomp.StompFrame)command;
+
+        if (stomp.getAction().equals(Stomp.Commands.KEEPALIVE)) {
+            os.write(Stomp.BREAK);
+            return;
+        }
+
+        StringBuilder builder = new StringBuilder();
+
+        os.write(marshalHeaders(stomp, builder).toString().getBytes("UTF-8"));
         os.write(stomp.getContent());
         os.write(END_OF_FRAME);
     }
 
+    public String marshalToString(StompFrame stomp) throws IOException {
+        if (stomp.getAction().equals(Stomp.Commands.KEEPALIVE)) {
+            return String.valueOf((char)Stomp.BREAK);
+        }
+
+        StringBuilder buffer = new StringBuilder();
+        marshalHeaders(stomp, buffer);
+
+        if (stomp.getContent() != null) {
+            String contentString = new String(stomp.getContent(), "UTF-8");
+            buffer.append(contentString);
+        }
+
+        buffer.append('\u0000');
+        return buffer.toString();
+    }
+
     @Override
     public Object unmarshal(DataInput in) throws IOException {
-
         try {
-
             // parse action
             String action = parseAction(in, frameSize);
 
@@ -212,7 +232,7 @@ public class StompWireFormat implements WireFormat {
     }
 
     protected HashMap<String, String> parseHeaders(DataInput in, AtomicLong frameSize) throws IOException {
-        HashMap<String, String> headers = new HashMap<String, String>(25);
+        HashMap<String, String> headers = new HashMap<>(25);
         while (true) {
             ByteSequence line = readHeaderLine(in, MAX_HEADER_LENGTH, "The maximum header length was exceeded");
             if (line != null && line.length > 1) {