[finagle-http] Preserve reader and writer state when constructing req…

…uest and response

Problem

The constructor for Responses and Requests that takes HttpResponse and
HttpRequest should perform a match to see if the parameter is already a
Response or Request, and in those casesjust return them. This is so that
the state of thereader and writer is preserved.

Solution

Return the param if it's already the correct type, and if we need to
create a new object, make sure to copy the reader and writer.

RB_ID=690768

finagle-http/src/main/scala/com/twitter/finagle/http/Request.scala

@@ -222,20 +222,32 @@ object Request {
     apply(new DefaultHttpRequest(version, method, uri))
 
   /** Create Request from HttpRequest. */
-  def apply(httpRequestArg: HttpRequest): MockRequest = {
-    new MockRequest {
-      override val httpRequest = httpRequestArg
-      override val httpMessage = httpRequestArg
-      override val remoteSocketAddress = new InetSocketAddress("127.0.0.1", 12345)
+  def apply(httpRequestArg: HttpRequest): MockRequest =
+    httpRequestArg match {
+      case req: MockRequest => req
+      case _ => new MockRequest {
+        override val httpRequest = httpRequestArg
+        override val httpMessage = httpRequestArg
+        override val remoteSocketAddress = new InetSocketAddress("127.0.0.1", 12345)
+      }
     }
-  }
 
   /** Create Request from HttpRequest and Channel.  Used by Codec. */
   def apply(httpRequestArg: HttpRequest, channel: Channel): Request =
-    new Request {
-      val httpRequest = httpRequestArg
-      override val httpMessage = httpRequestArg
-      lazy val remoteSocketAddress = channel.getRemoteAddress.asInstanceOf[InetSocketAddress]
+    httpRequestArg match {
+      case req: Request => new Request {
+        val httpRequest = req
+        override val httpMessage = req
+        override val remoteSocketAddress = channel.getRemoteAddress.asInstanceOf[InetSocketAddress]
+        override val reader = req.reader
+        override val writer = req.writer
+      }
+
+      case _ => new Request {
+        val httpRequest = httpRequestArg
+        override val httpMessage = httpRequestArg
+        override val remoteSocketAddress = channel.getRemoteAddress.asInstanceOf[InetSocketAddress]
+      }
     }
 
   // for testing

finagle-http/src/main/scala/com/twitter/finagle/http/Response.scala

@@ -83,8 +83,11 @@ object Response {
 
   /** Create Response from HttpResponse. */
   def apply(httpResponseArg: HttpResponse): Response =
-    new Response {
-      final val httpResponse = httpResponseArg
+    httpResponseArg match {
+      case res: Response => res
+      case _ => new Response {
+        final val httpResponse = httpResponseArg
+      }
     }
 
   /** Create Response from HttpRequest. */

finagle-http/src/test/scala/com/twitter/finagle/http/RequestTest.scala

@@ -1,5 +1,8 @@
 package com.twitter.finagle.http
 
+import com.twitter.conversions.time._
+import com.twitter.io.Buf
+import com.twitter.util.Await
 import org.jboss.netty.handler.codec.http.{DefaultHttpRequest, HttpMethod, HttpVersion}
 import org.junit.runner.RunWith
 import org.scalatest.FunSuite
@@ -25,6 +28,14 @@ class RequestTest extends FunSuite {
     }
   }
 
+  test("preserve stream in construction") {
+    val req = Request()
+    req.writer.write(Buf.Utf8("12")) before req.writer.close()
+    val f = Request(req).reader.read(1)
+    val g = Request(req).reader.read(1)
+    Await.result(f.join(g), 1.second) == (Buf.Utf8("1"), Buf.Utf8("2"))
+  }
+
   test("path") {
     val tests = Map(
       "/" -> "/",

finagle-http/src/test/scala/com/twitter/finagle/http/ResponseTest.scala

@@ -1,7 +1,9 @@
 package com.twitter.finagle.http
 
-import org.jboss.netty.handler.codec.http.{DefaultHttpRequest, DefaultHttpResponse, HttpMethod,
-  HttpResponseStatus, HttpVersion}
+import com.twitter.conversions.time._
+import com.twitter.io.Buf
+import com.twitter.util.Await
+import org.jboss.netty.handler.codec.http._
 import org.junit.runner.RunWith
 import org.scalatest.FunSuite
 import org.scalatest.junit.JUnitRunner
@@ -22,6 +24,14 @@ class ResponseTest extends FunSuite {
     }
   }
 
+  test("preserve stream in construction") {
+    val res = Response()
+    res.writer.write(Buf.Utf8("12")) before res.writer.close()
+    val f = Response(res).reader.read(1)
+    val g = Response(res).reader.read(1)
+    Await.result(f.join(g), 1.second) == (Buf.Utf8("1"), Buf.Utf8("2"))
+  }
+
   test("encode") {
     val response = Response()
     response.headers.set("Server", "macaw")