Fix thread deadlock issue in RPC handler.

scuttlebutt-rpc/build.gradle

@@ -4,6 +4,7 @@ dependencies {
   compile project(':bytes')
   compile project(':concurrent')
   compile project(':crypto')
+  compileOnly 'io.vertx:vertx-core'
   compile project(':scuttlebutt')
   compile project(':scuttlebutt-handshake')
   compile 'org.logl:logl-api'

scuttlebutt-rpc/src/main/java/net/consensys/cava/scuttlebutt/rpc/RPCMessage.java

@@ -15,11 +15,12 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import net.consensys.cava.bytes.Bytes;
+import net.consensys.cava.scuttlebutt.rpc.mux.exceptions.RPCRequestFailedException;
 
 import java.io.IOException;
+import java.util.Optional;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Optional;
 
 /**
  * Decoded RPC message, making elements of the message available directly.
@@ -104,16 +105,40 @@ public Optional<RPCErrorBody> getErrorBody(ObjectMapper objectMapper) {
 
     if (!isErrorMessage()) {
       // If the body of the response is 'true' or the error flag isn't set, it's a successful end condition
-      return Optional.absent();
+      return Optional.empty();
     } else {
       try {
         return Optional.of(asJSON(objectMapper, RPCErrorBody.class));
       } catch (IOException e) {
-        return Optional.absent();
+        return Optional.empty();
       }
     }
   }
 
+  /**
+   *
+   * @param objectMapper the objectmatter to deserialize the error with.
+   *
+   * @return an exception if this represents an error RPC response, otherwise nothing
+   */
+  public Optional<RPCRequestFailedException> getException(ObjectMapper objectMapper) {
+    if (isErrorMessage()) {
+      Optional<RPCRequestFailedException> exception =
+          getErrorBody(objectMapper).map(errorBody -> new RPCRequestFailedException(errorBody.getMessage()));
+
+      if (!exception.isPresent()) {
+        // If we failed to deserialize into the RPCErrorBody type there may be a bug in the server implementation
+        // which prevented it returning the correct type, so we just print whatever it returned
+        return Optional.of(new RPCRequestFailedException(this.asString()));
+      } else {
+        return exception;
+      }
+
+    } else {
+      return Optional.empty();
+    }
+  }
+
   /**
    * Provides the type of the body of the message: a binary message, a UTF-8 string or a JSON message.
    * 

scuttlebutt-rpc/src/main/java/net/consensys/cava/scuttlebutt/rpc/RPCResponse.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2019 ConsenSys AG.
+ *
+ * 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 net.consensys.cava.scuttlebutt.rpc;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import net.consensys.cava.bytes.Bytes;
+import net.consensys.cava.scuttlebutt.rpc.RPCFlag.BodyType;
+
+import java.io.IOException;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * A successful RPC response.
+ */
+public class RPCResponse {
+
+  private final Bytes body;
+  private final BodyType bodyType;
+
+  /**
+   * A successful RPC response.
+   *
+   * @param body the body of the response in bytes
+   * @param bodyType the type of the response (e.g. JSON, UTF-8 or binary.)
+   */
+  public RPCResponse(Bytes body, BodyType bodyType) {
+
+    this.body = body;
+    this.bodyType = bodyType;
+  }
+
+  /**
+   * @return the RPC response body
+   */
+  public Bytes body() {
+    return body;
+  }
+
+  /**
+   * @return The type of the data contained in the body.
+   */
+  public BodyType bodyType() {
+    return bodyType;
+  }
+
+  /**
+   * Provides the body of the message as a UTF-8 string.
+   *
+   * @return the body of the message as a UTF-8 string
+   */
+  public String asString() {
+    return new String(body().toArrayUnsafe(), UTF_8);
+  }
+
+  /**
+   * Provides the body of the message, marshalled as a JSON object.
+   *
+   * @param objectMapper the object mapper to deserialize with
+   * @param clazz the JSON object class
+   * @param <T> the matching JSON object class
+   * @return a new instance of the JSON object class
+   * @throws IOException if an error occurs during marshalling
+   */
+  public <T> T asJSON(ObjectMapper objectMapper, Class<T> clazz) throws IOException {
+    return objectMapper.readerFor(clazz).readValue(body().toArrayUnsafe());
+  }
+
+}

scuttlebutt-rpc/src/main/java/net/consensys/cava/scuttlebutt/rpc/mux/Multiplexer.java

@@ -14,7 +14,7 @@
 
 import net.consensys.cava.concurrent.AsyncResult;
 import net.consensys.cava.scuttlebutt.rpc.RPCAsyncRequest;
-import net.consensys.cava.scuttlebutt.rpc.RPCMessage;
+import net.consensys.cava.scuttlebutt.rpc.RPCResponse;
 import net.consensys.cava.scuttlebutt.rpc.RPCStreamRequest;
 import net.consensys.cava.scuttlebutt.rpc.mux.exceptions.ConnectionClosedException;
 
@@ -35,7 +35,7 @@ public interface Multiplexer {
    *
    * @return an async result which will be completed with the result or an error if the request fails.
    */
-  AsyncResult<RPCMessage> makeAsyncRequest(RPCAsyncRequest request);
+  AsyncResult<RPCResponse> makeAsyncRequest(RPCAsyncRequest request) throws JsonProcessingException;
 
   /**
    * Creates a request which opens a stream (e.g. a 'source' in the protocol docs.)