finagle-http: Handle last empty chunks more thoughtfully

Problem / Solution

We're not doing anything special about last empty chunks (both trailers and payload are empty)
although they are fairly popular in our pipelines. Let's handle them explicitly and save some
allocations.

JIRA Issues: CSL-7728

Differential Revision: https://phabricator.twitter.biz/D284623

finagle-base-http/src/main/scala/com/twitter/finagle/http/Chunk.scala

@@ -70,6 +70,11 @@ private[finagle] object Chunk {
     def isLast: Boolean = true
   }
 
+  /**
+   * A last (end of stream) empty [[Chunk]] that has neither a payload nor trailers.
+   */
+  val empty: Chunk = Last(Buf.Empty, HeaderMap.Empty)
+
   /**
    * Creates a non last [[Chunk]] that carries a payload, `bytes`.
    */

finagle-base-http/src/main/scala/com/twitter/finagle/http/EmptyHeaderMap.scala

@@ -0,0 +1,20 @@
+package com.twitter.finagle.http
+
+/**
+ * An empty and read-only [[HeaderMap]]. Use [[HeaderMap.Empty]] for a singleton instance.
+ */
+private final class EmptyHeaderMap extends HeaderMap {
+  def getAll(key: String): Seq[String] = Nil
+  def get(key: String): Option[String] = None
+  def add(k: String, v: String): HeaderMap = this
+  def addUnsafe(k: String, v: String): HeaderMap = this
+  def set(k: String, v: String): HeaderMap = this
+  def setUnsafe(k: String, v: String): HeaderMap = this
+
+  def +=(kv: (String, String)): this.type = this
+  def -=(key: String): this.type = this
+  def iterator: Iterator[(String, String)] = Iterator.empty
+
+  override def isEmpty: Boolean = true
+  override def toString: String = "EmptyHeaderMap"
+}

finagle-base-http/src/main/scala/com/twitter/finagle/http/HeaderMap.scala

@@ -84,6 +84,11 @@ abstract class HeaderMap
 
 object HeaderMap {
 
+  /**
+   * Empty, read-only [[HeaderMap]].
+   */
+  val Empty: HeaderMap = new EmptyHeaderMap
+
   /** Create a new HeaderMap from header list.
    *
    * @note the headers are added to the new `HeaderMap` via `add` operations.

finagle-netty4-http/src/main/scala/com/twitter/finagle/netty4/http/Netty4StreamTransport.scala

@@ -25,11 +25,15 @@ private[http] object Netty4StreamTransport {
       private def copyLoop(): Future[Unit] =
         trans.read().flatMap {
           case chunk: LastHttpContent =>
-            // TODO (vk): Optimize against LastHttpContent.EMPTY_LAST_CONTENT (very common case)
-            val last = Chunk.Last(
-              ByteBufConversion.byteBufAsBuf(chunk.content()),
-              Bijections.netty.headersToFinagle(chunk.trailingHeaders())
-            )
+            val last =
+              if (!chunk.content.isReadable && chunk.trailingHeaders().isEmpty)
+                Chunk.empty
+              else
+                Chunk.Last(
+                  ByteBufConversion.byteBufAsBuf(chunk.content()),
+                  Bijections.netty.headersToFinagle(chunk.trailingHeaders())
+                )
+
             pipe.write(last)
 
           case chunk: HttpContent =>
@@ -101,12 +105,19 @@ private[http] object Netty4StreamTransport {
     // We need to read one more time before writing trailers to ensure HTTP stream isn't malformed.
     def terminate(trailers: HeaderMap): Future[Unit] = r.read().flatMap {
       case None =>
-        // TODO (vk): PR against Netty; we need to construct out of given Headers
-        val lastHttpContent =
-          new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER, false /*validateHeaders*/ )
+        // TODO (vk): PR against Netty; we need to construct out of given Headers so we avoid
+        // copying afterwards.
+        val last =
+          if (trailers.isEmpty) LastHttpContent.EMPTY_LAST_CONTENT
+          else {
+            val content =
+              new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER, false /*validateHeaders*/ )
+
+            Bijections.finagle.writeFinagleHeadersToNetty(trailers, content.trailingHeaders())
+            content
+          }
 
-        Bijections.finagle.writeFinagleHeadersToNetty(trailers, lastHttpContent.trailingHeaders())
-        trans.write(lastHttpContent)
+        trans.write(last)
 
       case _ =>
         Future.exception(