Adding the low watermark option

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

@@ -136,6 +136,9 @@ akka {
       # (I&O) Sets the high water mark for the in and outbound sockets, set to 0b for platform default
       write-buffer-high-water-mark = 0b
 
+      # (I&O) Sets the low water mark for the in and outbound sockets, set to 0b for platform default
+      write-buffer-low-water-mark = 0b
+
       # (I&O) Sets the send buffer size of the Sockets, set to 0b for platform default
       send-buffer-size = 0b
 

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

@@ -150,6 +150,7 @@ private[akka] class ActiveRemoteClient private[akka] (
       settings.ReceiveBufferSize.foreach(sz ⇒ b.setOption("receiveBufferSize", sz))
       settings.SendBufferSize.foreach(sz ⇒ b.setOption("sendBufferSize", sz))
       settings.WriteBufferHighWaterMark.foreach(sz ⇒ b.setOption("writeBufferHighWaterMark", sz))
+      settings.WriteBufferLowWaterMark.foreach(sz ⇒ b.setOption("writeBufferLowWaterMark", sz))
       settings.OutboundLocalAddress.foreach(s ⇒ b.setOption("localAddress", new InetSocketAddress(s, 0)))
       bootstrap = b
 

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

@@ -48,6 +48,7 @@ private[akka] class NettyRemoteServer(val netty: NettyRemoteTransport) {
     settings.ReceiveBufferSize.foreach(sz ⇒ b.setOption("receiveBufferSize", sz))
     settings.SendBufferSize.foreach(sz ⇒ b.setOption("sendBufferSize", sz))
     settings.WriteBufferHighWaterMark.foreach(sz ⇒ b.setOption("writeBufferHighWaterMark", sz))
+    settings.WriteBufferLowWaterMark.foreach(sz ⇒ b.setOption("writeBufferLowWaterMark", sz))
     b
   }
 

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

@@ -48,6 +48,7 @@ private[akka] class NettySettings(config: Config, val systemName: String) {
   }
 
   val WriteBufferHighWaterMark: Option[Int] = optionSize("write-buffer-high-water-mark")
+  val WriteBufferLowWaterMark: Option[Int] = optionSize("write-buffer-low-water-mark")
   val SendBufferSize: Option[Int] = optionSize("send-buffer-size")
   val ReceiveBufferSize: Option[Int] = optionSize("receive-buffer-size")
 

akka-remote/src/test/scala/akka/remote/RemoteConfigSpec.scala

@@ -57,6 +57,7 @@ class RemoteConfigSpec extends AkkaSpec(
       AllTimeout must be(0 millis)
       ReconnectionTimeWindow must be(10 minutes)
       WriteBufferHighWaterMark must be(None)
+      WriteBufferLowWaterMark must be(None)
       SendBufferSize must be(None)
       ReceiveBufferSize must be(None)
     }