finagle-netty4: wire channeltransport into listener

now that we have a netty4 channel transport, let's use it.

bonus: make tcp_nodelay configurable.

RB_ID=772498

finagle-core/src/main/scala/com/twitter/finagle/server/Listener.scala

@@ -59,4 +59,19 @@ object Listener {
     implicit val param = Stack.Param(TrafficClass(None))
   }
 
+  /**
+   * Configures TCP_NODELAY on the server socket
+   * @param b `true` disables Nagle's algorithm.
+   * @note This param only applies to netty's >= 4.
+   *      Netty 3's socket options are still configured by
+   *      the [[com.twitter.finagle.netty3.Netty3Listener]] instance.
+   */
+  case class NoDelay(b: Boolean) {
+    def mk(): (NoDelay, Stack.Param[NoDelay]) =
+      (this, NoDelay.param)
+  }
+
+  object NoDelay {
+    implicit val param = Stack.Param(NoDelay(true))
+  }
 }

finagle-netty4/src/main/scala/com/twitter/finagle/netty4/Netty4Listener.scala

@@ -3,11 +3,10 @@ package com.twitter.finagle.netty4
 import com.twitter.concurrent.NamedPoolThreadFactory
 import com.twitter.finagle._
 import com.twitter.finagle.netty4.channel.{ServerBridge, Netty4ChannelInitializer}
-//import com.twitter.finagle.netty4.transport.SocketChannelTransport // CSL-2050
+import com.twitter.finagle.netty4.transport.ChannelTransport
 import com.twitter.finagle.server.Listener
 import com.twitter.finagle.ssl.Engine
 import com.twitter.finagle.transport.Transport
-import com.twitter.finagle.util.DefaultLogger
 import com.twitter.util._
 import io.netty.bootstrap.ServerBootstrap
 import io.netty.channel._
@@ -49,17 +48,18 @@ private[netty4] object PipelineInit {
  */
 case class Netty4Listener[In, Out](
     params: Stack.Params,
-    transportFactory: SocketChannel => Transport[In, Out] // CSL-2050 = new SocketChannelTransport[In, Out](_),
-    ) extends Listener[In, Out] {                         // todo: also, rm the transportFactory param
+    transportFactory: SocketChannel => Transport[In, Out] = new ChannelTransport[In, Out](_)
+  ) extends Listener[In, Out] {
 
   private[this] val PipelineInit(pipelineInit) = params[PipelineInit]
 
   // transport params
   private[this] val Transport.Liveness(_, _, keepAlive) = params[Transport.Liveness]
   private[this] val Transport.BufferSizes(sendBufSize, recvBufSize) = params[Transport.BufferSizes]
 
-  // listener params
+  // socket params
   private[this] val Listener.Backlog(backlog) = params[Listener.Backlog]
+  private[this] val Listener.NoDelay(noDelay) = params[Listener.NoDelay]
 
 
   /**
@@ -87,7 +87,7 @@ case class Netty4Listener[In, Out](
       val bootstrap = new ServerBootstrap()
       bootstrap.channel(classOf[NioServerSocketChannel])
       bootstrap.group(bossLoop, WorkerPool)
-      bootstrap.childOption[JBool](ChannelOption.TCP_NODELAY, true)
+      bootstrap.childOption[JBool](ChannelOption.TCP_NODELAY, noDelay)
       //bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) //todo: investigate pooled allocator CSL-2089
       //bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
       bootstrap.option[JBool](ChannelOption.SO_REUSEADDR, true)

finagle-netty4/src/test/scala/com/twitter/finagle/netty4/Netty4ListenerTest.scala

@@ -71,8 +71,7 @@ class Netty4ListenerTest extends FunSuite with Eventually with IntegrationPatien
   }
 
 
-  // ignore until CSL-2050
-  ignore("frames pipeline messages and bridges transports and service dispatchers (aka it works end-to-end)") {
+  test("frames pipeline messages and bridges transports and service dispatchers (aka it works end-to-end)") {
     val ctx = new StatsCtx { }
     import ctx._
 
@@ -85,10 +84,7 @@ class Netty4ListenerTest extends FunSuite with Eventually with IntegrationPatien
     }
 
     val p = Params.empty + Label("test") + Stats(sr) + PipelineInit(StringServerInit)
-    val listener = Netty4Listener[String, String](
-        p,
-        transportFactory = ??? // CSL-2050 new SocketChannelTransport[String, String](_),
-      )
+    val listener = Netty4Listener[String, String](p)
 
     @volatile var observedRequest: Option[String] = None