Added support for a config flag that allows the user to reuse the def…

…ault dispatcher as the ExecutorService for the Netty Remote Pipeline (IO)

akka-remote/src/main/resources/reference.conf

@@ -92,6 +92,10 @@ akka {
       # (I) Reuse inbound connections for outbound messages
       use-passive-connections = on
 
+      # (I) If "on" then the default dispatcher will be used to accept inbound connections,
+      # and IO. If "off" then dedicated threads will be used.
+      use-default-dispatcher-for-io = off
+
       # (I) The hostname or ip to bind the remoting to,
       # InetAddress.getLocalHost.getHostAddress is used if empty
       hostname = ""

akka-remote/src/main/scala/akka/remote/netty/Server.scala

@@ -26,9 +26,10 @@ class NettyRemoteServer(val netty: NettyRemoteTransport) {
 
   val ip = InetAddress.getByName(settings.Hostname)
 
-  private val factory = new NioServerSocketChannelFactory(
-    Executors.newCachedThreadPool(netty.system.threadFactory),
-    Executors.newCachedThreadPool(netty.system.threadFactory))
+  private val factory = {
+    val boss, worker = if (settings.UseDefaultDispatcherForIO) netty.system.dispatcher else Executors.newCachedThreadPool()
+    new NioServerSocketChannelFactory(boss, worker)
+  }
 
   private val executionHandler = new ExecutionHandler(netty.executor)
 

akka-remote/src/main/scala/akka/remote/netty/Settings.scala

@@ -27,6 +27,7 @@ class NettySettings(config: Config, val systemName: String) {
   }
 
   val UsePassiveConnections = getBoolean("use-passive-connections")
+  val UseDefaultDispatcherForIO = getBoolean("use-default-dispatcher-for-io")
 
   val ReconnectionTimeWindow = Duration(getMilliseconds("reconnection-time-window"), MILLISECONDS)
   val ReadTimeout = Duration(getMilliseconds("read-timeout"), MILLISECONDS)