Add MessagePack support for Java SignalR Client (#23532)

* Implement ParseMessages for java messagePack client

* Fix some spacing & syntax

* Implement write

* Tab -> Spaces

* MessagePacker -> MessageBufferPacker

* Tabs -> Spaces

* Tabs -> Spaces

* InvocationMessage may not include streamIDs

* Only 1 ctor per message type

* Fixup HubConnection.java

* Change return type of parseMessages to List

* Fix HubConnection

* Check for primitive value before returning

* Implement length header prefix

* Minor fixes

* Use ByteBuffer to read length header

* Add case for Char

* Close unpacker

* Typo

* Override onMessage w/ ByteString

* Change OKHttpWebSocketWrapper

* Account for nil InvocationId

* Change interface & MessagePack impl

* Update JsonHubProtocol

* Use ByteBuffer

* Fixup HubConnection

* Fixup more stuff

* Convert more stuff to ByteBuffer

* Account for ReadOnly

* Spacing

* No need to reset ByteBuffer when setting position

* Add Protocol to HubConnection ctor

* Set default, make stuff public

* Fixup tests

* More test cleanup

* Spacing

* only grab remaining buffer bytes in json

* Last test fixes

* Get rid of some unused imports

* First round of msgpack tests

* Flip condition

* Respond to feedback

* Spacing

* More tests

* Add test for primitives

* Add more tests, start using msgpack-jackson

* Fix build.gradle

* Remove debug prints

* Start using Type instead of Class

* Add overloads for Type, make messagePack readValue() more efficient

* Apply feedback, add some tests

* Add some tests, fix some tests

* Fix tests for real

* Add a whole buncha tests

* Add TestUtils change that I didn't commit yesterday

* Respond to some feedback

* Add a couple Json tests

* Apply more feedback

* Move readonly fix to msgpack

* Minor optimization

* Fixup some javadocs

* Respond to feedback

* Remove TypeReference, make Protocols private again

* Feedback

Author: wtgodbe

src/SignalR/clients/java/signalr/build.gradle

@@ -39,6 +39,8 @@ dependencies {
     implementation 'com.squareup.okhttp3:okhttp:3.11.0'
     api 'io.reactivex.rxjava2:rxjava:2.2.3'
     implementation 'org.slf4j:slf4j-api:1.7.25'
+    compile 'org.msgpack:msgpack-core:0.8.20'
+    compile 'org.msgpack:jackson-dataformat-msgpack:0.8.20'
 }
 
 spotless {

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

@@ -3,6 +3,7 @@
 
 package com.microsoft.signalr;
 
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -13,10 +14,10 @@ class CallbackMap {
     private final Map<String, List<InvocationHandler>> handlers = new HashMap<>();
     private final ReentrantLock lock = new ReentrantLock();
 
-    public InvocationHandler put(String target, ActionBase action, Class<?>... classes) {
+    public InvocationHandler put(String target, ActionBase action, Type... types) {
         try {
             lock.lock();
-            InvocationHandler handler = new InvocationHandler(action, classes);
+            InvocationHandler handler = new InvocationHandler(action, types);
             if (!handlers.containsKey(target)) {
                 handlers.put(target, new ArrayList<>());
             }

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

@@ -3,14 +3,28 @@
 
 package com.microsoft.signalr;
 
+import java.util.Map;
+
 final class CancelInvocationMessage extends HubMessage {
     private final int type = HubMessageType.CANCEL_INVOCATION.value;
+    private Map<String, String> headers;
     private final String invocationId;
-
-    public CancelInvocationMessage(String invocationId) {
+    
+    public CancelInvocationMessage(Map<String, String> headers, String invocationId) {
+        if (headers != null && !headers.isEmpty()) {
+            this.headers = headers;
+        }
         this.invocationId = invocationId;
     }
 
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
+    
+    public String getInvocationId() {
+        return invocationId;
+    }
+    
     @Override
     public HubMessageType getMessageType() {
         return HubMessageType.CANCEL_INVOCATION;

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

@@ -5,21 +5,35 @@
 
 final class CloseMessage extends HubMessage {
     private final String error;
+    private final boolean allowReconnect;
 
     @Override
     public HubMessageType getMessageType() {
         return HubMessageType.CLOSE;
     }
 
     public CloseMessage() {
-        this(null);
+        this(null, false);
     }
 
     public CloseMessage(String error) {
+        this(error, false);
+    }
+    
+    public CloseMessage(boolean allowReconnect) {
+        this(null, allowReconnect);
+    }
+
+    public CloseMessage(String error, boolean allowReconnect) {
         this.error = error;
+        this.allowReconnect = allowReconnect;
     }
 
     public String getError() {
         return this.error;
     }
+    
+    public boolean getAllowReconnect() {
+        return this.allowReconnect;
+    }
 }

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

@@ -3,20 +3,30 @@
 
 package com.microsoft.signalr;
 
+import java.util.Map;
+
 final class CompletionMessage extends HubMessage {
     private final int type = HubMessageType.COMPLETION.value;
+    private Map<String, String> headers;
     private final String invocationId;
     private final Object result;
     private final String error;
-
-    public CompletionMessage(String invocationId, Object result, String error) {
+    
+    public CompletionMessage(Map<String, String> headers, String invocationId, Object result, String error) {
+        if (headers != null && !headers.isEmpty()) {
+            this.headers = headers;
+        }
         if (error != null && result != null) {
             throw new IllegalArgumentException("Expected either 'error' or 'result' to be provided, but not both.");
         }
         this.invocationId = invocationId;
         this.result = result;
         this.error = error;
     }
+    
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
 
     public Object getResult() {
         return result;

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

@@ -4,6 +4,7 @@
 package com.microsoft.signalr;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -15,6 +16,7 @@
 import io.reactivex.Single;
 import io.reactivex.subjects.SingleSubject;
 import okhttp3.*;
+import okio.ByteString;
 
 final class DefaultHttpClient extends HttpClient {
     private OkHttpClient client = null;
@@ -104,7 +106,7 @@ public Single<HttpResponse> send(HttpRequest httpRequest) {
     }
 
     @Override
-    public Single<HttpResponse> send(HttpRequest httpRequest, String bodyContent) {
+    public Single<HttpResponse> send(HttpRequest httpRequest, ByteBuffer bodyContent) {
         Request.Builder requestBuilder = new Request.Builder().url(httpRequest.getUrl());
 
         switch (httpRequest.getMethod()) {
@@ -114,7 +116,7 @@ public Single<HttpResponse> send(HttpRequest httpRequest, String bodyContent) {
             case "POST":
                 RequestBody body;
                 if (bodyContent != null) {
-                    body = RequestBody.create(MediaType.parse("text/plain"), bodyContent);
+                    body = RequestBody.create(MediaType.parse("text/plain"), ByteString.of(bodyContent));
                 } else {
                     body = RequestBody.create(null, new byte[]{});
                 }
@@ -150,7 +152,7 @@ public void onFailure(Call call, IOException e) {
             @Override
             public void onResponse(Call call, Response response) throws IOException {
                 try (ResponseBody body = response.body()) {
-                    HttpResponse httpResponse = new HttpResponse(response.code(), response.message(), body.string());
+                    HttpResponse httpResponse = new HttpResponse(response.code(), response.message(), ByteBuffer.wrap(body.bytes()));
                     responseSubject.onSuccess(httpResponse);
                 }
             }

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

@@ -3,15 +3,18 @@
 
 package com.microsoft.signalr;
 
+import java.nio.charset.StandardCharsets;
+import java.nio.ByteBuffer;
+
 import com.google.gson.Gson;
 
 final class HandshakeProtocol {
     private static final Gson gson = new Gson();
     private static final String RECORD_SEPARATOR = "\u001e";
 
-    public static String createHandshakeRequestMessage(HandshakeRequestMessage message) {
+    public static ByteBuffer createHandshakeRequestMessage(HandshakeRequestMessage message) {
         // The handshake request is always in the JSON format
-        return gson.toJson(message) + RECORD_SEPARATOR;
+        return ByteBuffer.wrap((gson.toJson(message) + RECORD_SEPARATOR).getBytes(StandardCharsets.UTF_8));
     }
 
     public static HandshakeResponseMessage parseHandshakeResponse(String message) {

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

@@ -3,6 +3,7 @@
 
 package com.microsoft.signalr;
 
+import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -45,23 +46,23 @@ public Map<String, String> getHeaders() {
 class HttpResponse {
     private final int statusCode;
     private final String statusText;
-    private final String content;
+    private final ByteBuffer content;
 
     public HttpResponse(int statusCode) {
         this(statusCode, "");
     }
 
     public HttpResponse(int statusCode, String statusText) {
-        this(statusCode, statusText, "");
+        this(statusCode, statusText, ByteBuffer.wrap(new byte[] {}));
     }
 
-    public HttpResponse(int statusCode, String statusText, String content) {
+    public HttpResponse(int statusCode, String statusText, ByteBuffer content) {
         this.statusCode = statusCode;
         this.statusText = statusText;
         this.content = content;
     }
 
-    public String getContent() {
+    public ByteBuffer getContent() {
         return content;
     }
 
@@ -95,7 +96,7 @@ public Single<HttpResponse> post(String url) {
         return this.send(request);
     }
 
-    public Single<HttpResponse> post(String url, String body, HttpRequest options) {
+    public Single<HttpResponse> post(String url, ByteBuffer body, HttpRequest options) {
         options.setUrl(url);
         options.setMethod("POST");
         return this.send(options, body);
@@ -122,7 +123,7 @@ public Single<HttpResponse> delete(String url, HttpRequest options) {
 
     public abstract Single<HttpResponse> send(HttpRequest request);
 
-    public abstract Single<HttpResponse> send(HttpRequest request, String body);
+    public abstract Single<HttpResponse> send(HttpRequest request, ByteBuffer body);
 
     public abstract WebSocketWrapper createWebSocket(String url, Map<String, String> headers);
 

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

@@ -16,6 +16,7 @@ public class HttpHubConnectionBuilder {
     private final String url;
     private Transport transport;
     private HttpClient httpClient;
+    private HubProtocol protocol = new JsonHubProtocol();
     private boolean skipNegotiate;
     private Single<String> accessTokenProvider;
     private long handshakeResponseTimeout = 0;
@@ -54,6 +55,16 @@ HttpHubConnectionBuilder withHttpClient(HttpClient httpClient) {
         this.httpClient = httpClient;
         return this;
     }
+    
+    /**
+     * Sets MessagePack as the {@link HubProtocol} to be used by the {@link HubConnection}.
+     *
+     * @return This instance of the HttpHubConnectionBuilder.
+     */
+    public HttpHubConnectionBuilder withMessagePackHubProtocol() {
+        this.protocol = new MessagePackHubProtocol();
+        return this;
+    }
 
     /**
      * Indicates to the {@link HubConnection} that it should skip the negotiate process.
@@ -133,7 +144,7 @@ public HttpHubConnectionBuilder setHttpClientBuilderCallback(Action1<OkHttpClien
      * @return A new instance of {@link HubConnection}.
      */
     public HubConnection build() {
-        return new HubConnection(url, transport, skipNegotiate, httpClient, accessTokenProvider,
+        return new HubConnection(url, transport, skipNegotiate, httpClient, protocol, accessTokenProvider,
             handshakeResponseTimeout, headers, transportEnum, configureBuilder);
     }
 }