diff --git a/scuttlebutt-rpc/src/main/java/org/apache/tuweni/scuttlebutt/rpc/mux/RPCHandler.java b/scuttlebutt-rpc/src/main/java/org/apache/tuweni/scuttlebutt/rpc/mux/RPCHandler.java
index b0f3639..eb3fadc 100644
--- a/scuttlebutt-rpc/src/main/java/org/apache/tuweni/scuttlebutt/rpc/mux/RPCHandler.java
+++ b/scuttlebutt-rpc/src/main/java/org/apache/tuweni/scuttlebutt/rpc/mux/RPCHandler.java
@@ -99,6 +99,7 @@
 
         awaitingAsyncResponse.put(requestNumber, result);
         Bytes bytes = RPCCodec.encodeRequest(message.body(), requestNumber, request.getRPCFlags());
+        logOutgoingRequest(message);
         sendBytes(bytes);
       }
     };
@@ -122,7 +123,16 @@
       Bytes requestBytes = RPCCodec.encodeRequest(message.body(), requestNumber, rpcFlags);
 
       Runnable closeStreamHandler = () -> {
-        endStream(requestNumber);
+
+        // Run on vertx context because this callback may be called from a different
+        // thread by the caller
+        vertx.runOnContext(new Handler<Void>() {
+          @Override
+          public void handle(Void event) {
+            endStream(requestNumber);
+          }
+        });
+
       };
 
       ScuttlebuttStreamHandler scuttlebuttStreamHandler = responseSink.apply(closeStreamHandler);
@@ -131,6 +141,7 @@
         scuttlebuttStreamHandler.onStreamError(new ConnectionClosedException());
       } else {
         streams.put(requestNumber, scuttlebuttStreamHandler);
+        logOutgoingRequest(message);
         sendBytes(requestBytes);
       }
 
@@ -140,6 +151,15 @@
     vertx.runOnContext(synchronizedRequest);
   }
 
+  private void logOutgoingRequest(RPCMessage rpcMessage) {
+    if (logger.isDebugEnabled()) {
+      String requestString = new String(rpcMessage.asString());
+      String logMessage = String.format("[%d] Outgoing request: %s", rpcMessage.requestNumber(), requestString);
+      logger.debug(logMessage);
+    }
+
+  }
+
   @Override
   public void close() {
     vertx.runOnContext((event) -> {
@@ -192,14 +212,14 @@
   private void handleRequest(RPCMessage rpcMessage) {
     // Not yet implemented
     logger.warn("Received incoming request, but we do not yet handle any requests: " + rpcMessage.asString());
-
   }
 
   private void handleResponse(RPCMessage response) {
     int requestNumber = response.requestNumber() * -1;
 
     if (logger.isDebugEnabled()) {
-      logger.debug("Incoming response: " + response.asString());
+      String logMessage = String.format("[%d] incoming response: %s", requestNumber, response.asString());
+      logger.debug(logMessage);
     }
 
     byte rpcFlags = response.rpcFlags();
@@ -214,9 +234,8 @@
       if (scuttlebuttStreamHandler != null) {
 
         if (response.isSuccessfulLastMessage()) {
-          streams.remove(requestNumber);
+          // Confirm our end of the stream close and inform the consumer of the stream that it is closed
           endStream(requestNumber);
-          scuttlebuttStreamHandler.onStreamEnd();
         } else if (exception.isPresent()) {
           scuttlebuttStreamHandler.onStreamError(exception.get());
         } else {
@@ -260,10 +279,32 @@
     messageSender.accept(bytes);
   }
 
+  /**
+   * Sends an stream close message over the RPC channel to for the given request number if we have not already closed
+   * our end of the stream.
+   *
+   * Removes the stream handler from the state, so any newly incoming messages until the other side of the stream has
+   * closed its end will be ignored.
+   *
+   * @param requestNumber the request number of the stream to send a close message over RPC for
+   */
   private void endStream(int requestNumber) {
     try {
-      Bytes streamEnd = RPCCodec.encodeStreamEndRequest(requestNumber);
-      sendBytes(streamEnd);
+      ScuttlebuttStreamHandler streamHandler = streams.remove(requestNumber);
+
+      // Only send the message if the stream hasn't already been closed at our end
+      if (streamHandler != null) {
+        Bytes streamEnd = RPCCodec.encodeStreamEndRequest(requestNumber);
+
+        streamHandler.onStreamEnd();
+        if (logger.isDebugEnabled()) {
+          String logMessage = String.format("[%d] Sending close stream message.", requestNumber);
+          logger.debug(logMessage);
+        }
+
+        sendBytes(streamEnd);
+      }
+
     } catch (JsonProcessingException e) {
       logger.warn("Unexpectedly could not encode stream end message to JSON.");
     }