Add control for if you want to use mbus threas for compression/encoding or your own.

2 files changed, 17 insertions, 1 deletions

diff --git a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java
index fdd72c12532..ac28f0247a2 100644
--- a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java
+++ b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetwork.java
@@ -56,6 +56,7 @@ public class RPCNetwork implements Network, MethodHandler {
     private static final Logger log = Logger.getLogger(RPCNetwork.class.getName());
     private final AtomicBoolean destroyed = new AtomicBoolean(false);
     private final Identity identity;
+    private final boolean useOwnThread;
     private final Supervisor orb;
     private final RPCTargetPool targetPool;
     private final RPCServicePool servicePool;
@@ -83,6 +84,7 @@ public class RPCNetwork implements Network, MethodHandler {
     public RPCNetwork(RPCNetworkParams params, SlobrokConfigSubscriber slobrokConfig) {
         this.slobroksConfig = slobrokConfig;
         identity = params.getIdentity();
+        useOwnThread = params.getSendInOwnThread();
         orb = new Supervisor(new Transport());
         orb.setMaxInputBufferSize(params.getMaxInputBufferSize());
         orb.setMaxOutputBufferSize(params.getMaxOutputBufferSize());
@@ -257,7 +259,12 @@ public class RPCNetwork implements Network, MethodHandler {
                     String.format("An error occurred while resolving version of recipient(s) [%s] from host '%s'.",
                                   buildRecipientListString(ctx), identity.getHostname()));
         } else {
-            executor.execute(new SendTask(owner.getProtocol(ctx.msg.getProtocol()), ctx));
+            SendTask sendTask = new SendTask(owner.getProtocol(ctx.msg.getProtocol()), ctx);
+            if ( ! useOwnThread ) {
+                executor.execute(sendTask);
+            } else {
+                sendTask.run();
+            }
         }
     }
 
diff --git a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java
index e50670530d3..ff3336ac37e 100755
--- a/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java
+++ b/messagebus/src/main/java/com/yahoo/messagebus/network/rpc/RPCNetworkParams.java
@@ -15,6 +15,7 @@ public class RPCNetworkParams {
     private Identity identity = new Identity("");
     private String slobrokConfigId = "admin/slobrok.0";
     private SlobroksConfig slobroksConfig = null;
+    private boolean sendInOwnThread = false;
     private int listenPort = 0;
     private int maxInputBufferSize = 256 * 1024;
     private int maxOutputBufferSize = 256 * 1024;
@@ -34,6 +35,7 @@ public class RPCNetworkParams {
      */
     public RPCNetworkParams(RPCNetworkParams params) {
         identity = new Identity(params.identity);
+        sendInOwnThread = params.sendInOwnThread;
         slobrokConfigId = params.slobrokConfigId;
         slobroksConfig = params.slobroksConfig;
         listenPort = params.listenPort;
@@ -62,6 +64,13 @@ public class RPCNetworkParams {
         return this;
     }
 
+    public boolean getSendInOwnThread() { return sendInOwnThread; }
+
+    public RPCNetworkParams setSendInOwnThread(boolean sendInOwnThread) {
+        this.sendInOwnThread = sendInOwnThread;
+        return this;
+    }
+
     /**
      * Returns the config id of the slobrok config.
      *