Convert more stuff to ByteBuffer

Author: wtgodbe

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/HubConnection.java

@@ -28,7 +28,7 @@
  * A connection used to invoke hub methods on a SignalR Server.
  */
 public class HubConnection implements AutoCloseable {
-    private static final String RECORD_SEPARATOR = "\u001e";
+    private static final byte RECORD_SEPARATOR = 0x1e;
     private static final List<Class<?>> emptyArray = new ArrayList<>();
     private static final int MAX_NEGOTIATE_ATTEMPTS = 100;
 
@@ -136,7 +136,7 @@ Transport getTransport() {
         }
 
         this.baseUrl = url;
-        this.protocol = new JsonHubProtocol();
+        this.protocol = new MessagePackHubProtocol();
 
         if (accessTokenProvider != null) {
             this.accessTokenProvider = accessTokenProvider;
@@ -167,8 +167,10 @@ Transport getTransport() {
         this.callback = (payload) -> {
             resetServerTimeout();
             if (!handshakeReceived) {
-                int handshakeLength = payload.indexOf(RECORD_SEPARATOR) + 1;
-                String handshakeResponseString = payload.substring(0, handshakeLength - 1);
+            	// The handshake will always be a UTF8 Json string, so we can convert the ByteBuffer to that to read the beginning of the payload
+            	String payloadStr = new String(payload.array(), StandardCharsets.UTF_8);
+                int handshakeLength = payloadStr.indexOf(RECORD_SEPARATOR) + 1;
+                String handshakeResponseString = payloadStr.substring(0, handshakeLength - 1);
                 HandshakeResponseMessage handshakeResponse;
                 try {
                     handshakeResponse = HandshakeProtocol.parseHandshakeResponse(handshakeResponseString);
@@ -186,15 +188,18 @@ Transport getTransport() {
                 }
                 handshakeReceived = true;
                 handshakeResponseSubject.onComplete();
+                
+                // Increment the ByteBuffer payload by the byte length of the handshake + the byte length of the record separator (1)
+                int readBytes = handshakeResponseString.getBytes(StandardCharsets.UTF_8).length + 1;
+                payload = payload.position(payload.position() + readBytes);
 
-                payload = payload.substring(handshakeLength);
                 // The payload only contained the handshake response so we can return.
-                if (payload.length() == 0) {
+                if (!payload.hasRemaining()) {
                     return;
                 }
             }
 
-            List<HubMessage> messages = protocol.parseMessages(ByteBuffer.wrap(payload.getBytes(StandardCharsets.UTF_8)), connectionState);
+            List<HubMessage> messages = protocol.parseMessages(payload, connectionState);
 
             for (HubMessage message : messages) {
                 logger.debug("Received message of type {}.", message.getMessageType());

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/JsonHubProtocol.java

@@ -39,7 +39,7 @@ public TransferFormat getTransferFormat() {
 
     @Override
     public List<HubMessage> parseMessages(ByteBuffer payload, InvocationBinder binder) {
-    	String payloadStr = new String(payload.array(), StandardCharsets.UTF_8);
+    String payloadStr = new String(payload.array(), StandardCharsets.UTF_8);
         if (payloadStr.length() == 0) {
             return null;
         }

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/LongPollingTransport.java

@@ -4,6 +4,7 @@
 package com.microsoft.signalr;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -101,7 +102,7 @@ private Completable poll(String url) {
                     } else {
                         if (response.getContent() != null) {
                             logger.debug("Message received.");
-                            onReceiveThread.submit(() ->this.onReceive(response.getContent()));
+                            onReceiveThread.submit(() ->this.onReceive(ByteBuffer.wrap(response.getContent().getBytes(StandardCharsets.UTF_8))));
                         } else {
                             logger.debug("Poll timed out, reissuing.");
                         }
@@ -139,7 +140,7 @@ public void setOnReceive(OnReceiveCallBack callback) {
     }
 
     @Override
-    public void onReceive(String message) {
+    public void onReceive(ByteBuffer message) {
         this.onReceiveCallBack.invoke(message);
         logger.debug("OnReceived callback has been invoked.");
     }

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/MessagePackHubProtocol.java

@@ -90,12 +90,12 @@ public List<HubMessage> parseMessages(ByteBuffer payload, InvocationBinder binde
                         break;
                 }
                 // Make sure that we actually read the right number of bytes
-                long readBytes = unpacker.getTotalReadBytes();
+                int readBytes = (int) unpacker.getTotalReadBytes();
                 if (readBytes != length) {
-                	throw new RuntimeException(String.format("MessagePack message was length %d but claimed to be length %d.", readBytes, length));
+                    throw new RuntimeException(String.format("MessagePack message was length %d but claimed to be length %d.", readBytes, length));
                 }
                 unpacker.close();
-                payload = payload.position(payload.position() + (int) readBytes);
+                payload = payload.position(payload.position() + readBytes);
             }
         } catch (MessagePackException | IOException ex) {
             throw new RuntimeException("Error reading MessagePack data.", ex);
@@ -160,7 +160,7 @@ private HubMessage createInvocationMessage(MessageUnpacker unpacker, InvocationB
         // invocationId may be nil
         String invocationId = null;
         if (!unpacker.tryUnpackNil()) {
-        	invocationId = unpacker.unpackString();
+            invocationId = unpacker.unpackString();
         }
 
         // For MsgPack, we represent an empty invocation ID as an empty string,

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/OkHttpWebSocketWrapper.java

@@ -64,7 +64,7 @@ public Completable stop() {
 
     @Override
     public Completable send(ByteBuffer message) {
-    	ByteString bs = ByteString.of(message);
+        ByteString bs = ByteString.of(message);
         websocketClient.send(bs);
         return Completable.complete();
     }
@@ -87,12 +87,12 @@ public void onOpen(WebSocket webSocket, Response response) {
 
         @Override
         public void onMessage(WebSocket webSocket, String message) {
-            onReceive.invoke(message);
+            onReceive.invoke(ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8)));
         }
 
         @Override
         public void onMessage(WebSocket webSocket, ByteString bytes) {
-        	onReceive.invoke(new String(bytes.toByteArray(), StandardCharsets.ISO_8859_1));
+            onReceive.invoke(bytes.asByteBuffer());
         }
 
         @Override

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/OnReceiveCallBack.java

@@ -3,6 +3,8 @@
 
 package com.microsoft.signalr;
 
+import java.nio.ByteBuffer;
+
 interface OnReceiveCallBack {
-    void invoke(String message);
+    void invoke(ByteBuffer message);
 }

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/Transport.java

@@ -11,7 +11,7 @@ interface Transport {
     Completable start(String url);
     Completable send(ByteBuffer message);
     void setOnReceive(OnReceiveCallBack callback);
-    void onReceive(String message);
+    void onReceive(ByteBuffer message);
     void setOnClose(TransportOnClosedCallback onCloseCallback);
     Completable stop();
 }

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/WebSocketTransport.java

@@ -72,7 +72,7 @@ public void setOnReceive(OnReceiveCallBack callback) {
     }
 
     @Override
-    public void onReceive(String message) {
+    public void onReceive(ByteBuffer message) {
         this.onReceiveCallBack.invoke(message);
     }