Remove extra buffer copy in StompNIOSSLTransport (#2107) (#2109)

Optimizes the StompNIOSSLTransport by removing the unnecessary extra
byte array allocation and buffer copy when processing frames

Author: cshannon

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

@@ -17,6 +17,7 @@
 package org.apache.activemq.transport.stomp;
 
 import java.io.ByteArrayInputStream;
+import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
@@ -47,11 +48,11 @@ public StompCodec(TcpTransport transport) {
         this.wireFormat = (StompWireFormat) Objects.requireNonNull(transport.getWireFormat());
     }
 
-    public void parse(ByteArrayInputStream input, int readSize) throws Exception {
+    public void parse(ByteBuffer input, int readSize) throws Exception {
        int i = 0;
        int b;
        while(i++ < readSize) {
-           b = input.read();
+           b = input.get();
            // skip repeating nulls
            if (!processedHeaders && previousByte == 0 && b == 0) {
                continue;

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

@@ -74,10 +74,7 @@ protected void initializeStreams() throws IOException {
 
     @Override
     protected void processCommand(ByteBuffer plain) throws Exception {
-        byte[] fill = new byte[plain.remaining()];
-        plain.get(fill);
-        ByteArrayInputStream input = new ByteArrayInputStream(fill);
-        codec.parse(input, fill.length);
+        codec.parse(plain, plain.remaining());
     }
 
     @Override

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

@@ -129,12 +129,8 @@ private void serviceRead() {
 
     protected void processBuffer(ByteBuffer buffer, int readSize) throws Exception {
         receiveCounter.addAndGet(readSize);
-
         buffer.flip();
-
-        ByteArrayInputStream input = new ByteArrayInputStream(buffer.array());
-        codec.parse(input, readSize);
-
+        codec.parse(buffer, readSize);
         // clear the buffer
         buffer.clear();
     }