Merge commit 'c4241c0c80534dbc249e7e15770b7f68341128b4'

Author: jenkins

CHANGELOG.rst

@@ -7,6 +7,14 @@ Note that ``RB_ID=#`` and ``PHAB_ID=#`` correspond to associated message in comm
 Unreleased
 ----------
 
+19.5.1
+------
+Fixed
+~~~~~
+
+* finatra: The added `c.t.finatra.http.RouteHint` was missing from the test-jar sources and has
+  been added. ``PHAB_ID=D317282``
+
 19.5.0
 ------
 

build.sbt

@@ -4,7 +4,7 @@ import scoverage.ScoverageKeys
 concurrentRestrictions in Global += Tags.limit(Tags.Test, 1)
 
 // All Twitter library releases are date versioned as YY.MM.patch
-val releaseVersion = "19.5.0"
+val releaseVersion = "19.5.1"
 
 lazy val buildSettings = Seq(
   version := releaseVersion,
@@ -627,6 +627,7 @@ lazy val httpTestJarSources =
     "com/twitter/finatra/http/HttpMockResponses",
     "com/twitter/finatra/http/HttpTest",
     "com/twitter/finatra/http/JsonAwareEmbeddedHttpClient",
+    "com/twitter/finatra/http/RouteHint",
     "com/twitter/finatra/http/StreamingJsonTestHelper")
 lazy val http = project
   .settings(projectSettings)

http/src/main/scala/com/twitter/finatra/http/internal/marshalling/CallbackConverter.scala

@@ -3,17 +3,20 @@ package com.twitter.finatra.http.internal.marshalling
 import com.twitter.concurrent.AsyncStream
 import com.twitter.finagle.http._
 import com.twitter.finatra.http.internal.marshalling.CallbackConverter.url
-import com.twitter.finatra.http.response.{ResponseBuilder, StreamingResponse}
-import com.twitter.finatra.http.streaming.{FromReader, StreamingRequest}
+import com.twitter.finatra.http.response.{
+  ResponseBuilder,
+  StreamingResponse => DeprecatedStreamingResponse
+}
+import com.twitter.finatra.http.streaming.{FromReader, StreamingRequest, StreamingResponse}
 import com.twitter.finatra.json.FinatraObjectMapper
 import com.twitter.finatra.json.internal.streaming.JsonStreamParser
 import com.twitter.inject.TypeUtils
 import com.twitter.io.{Buf, Reader}
 import com.twitter.util.{Future, FuturePool, Promise}
 import javax.inject.Inject
 import scala.concurrent.{ExecutionContext => ScalaExecutionContext, Future => ScalaFuture}
-import scala.util.{Failure, Success}
 import scala.reflect.runtime.universe._
+import scala.util.{Failure, Success}
 
 private object CallbackConverter {
   val url =
@@ -53,7 +56,7 @@ private[http] class CallbackConverter @Inject()(
     manifestRequestType match {
       case request if request == manifest[Request] =>
         callback.asInstanceOf[Request => ResponseType]
-      case streamingRequest if runtimeClassEqs[StreamingRequest[_, _]](streamingRequest) =>
+      case streamingRequest if runtimeClassEqs[StreamingRequest[Any, _]](streamingRequest) =>
         val streamIdentity = streamingRequest.typeArguments.head
         val streamType = streamingRequest.typeArguments.last
         request: Request =>
@@ -170,7 +173,8 @@ private[http] class CallbackConverter @Inject()(
   private def isStreamingType(manifested: Manifest[_]): Boolean = {
     runtimeClassEqs[AsyncStream[_]](manifested) ||
     runtimeClassEqs[Reader[_]](manifested) ||
-    runtimeClassEqs[StreamingResponse[_, _]](manifested)
+    runtimeClassEqs[DeprecatedStreamingResponse[_, _]](manifested) ||
+    runtimeClassEqs[StreamingResponse[Any, _]](manifested)
   }
 
   /**
@@ -185,6 +189,7 @@ private[http] class CallbackConverter @Inject()(
     requestCallback: Request => ResponseType,
     manifested: Manifest[_]
   ): Request => Future[Response] = {
+    // If the requestType is stream of Buf, pass it through without adding json fmt
     val (jsonPrefix, jsonSeparator, jsonSuffix) = {
       val typeArgs = TypeUtils.asManifest[RequestType].typeArguments
       if (typeArgs.size == 1 && typeArgs.head.runtimeClass == classOf[Buf]) {
@@ -197,37 +202,51 @@ private[http] class CallbackConverter @Inject()(
       case asyncStreamBuf if asyncStreamBuf == manifest[AsyncStream[Buf]] =>
         request: Request =>
           val asyncStream = requestCallback(request).asInstanceOf[AsyncStream[Buf]]
-          val streamingResponse =
-            StreamingResponse.apply[Buf](toBuf = a => a)(asyncStream = asyncStream)
-          streamingResponse.toFutureFinagleResponse
+          val depStreamingResponse =
+            DeprecatedStreamingResponse.apply[Buf](toBuf = a => a)(asyncStream = asyncStream)
+          depStreamingResponse.toFutureFinagleResponse
       case asyncStream if runtimeClassEqs[AsyncStream[_]](asyncStream) =>
         request: Request =>
           val asyncStream = requestCallback(request).asInstanceOf[AsyncStream[_]]
-          val streamingResponse =
-            StreamingResponse.apply(
+          val depStreamingResponse =
+            DeprecatedStreamingResponse.apply(
               toBuf = mapper.writeValueAsBuf,
               prefix = jsonPrefix,
               separator = jsonSeparator,
               suffix = jsonSuffix)(asyncStream = asyncStream)
-          streamingResponse.toFutureFinagleResponse
+          depStreamingResponse.toFutureFinagleResponse
       case readerBuf if readerBuf == manifest[Reader[Buf]] =>
         request: Request =>
           val reader = requestCallback(request).asInstanceOf[Reader[Buf]]
-          val streamingResponse =
-            StreamingResponse.apply[Buf](toBuf = a => a)(asyncStream = Reader.toAsyncStream(reader))
-          streamingResponse.toFutureFinagleResponse
+          val depStreamingResponse =
+            DeprecatedStreamingResponse.apply[Buf](toBuf = a => a)(
+              asyncStream = Reader.toAsyncStream(reader))
+          depStreamingResponse.toFutureFinagleResponse
       case reader if runtimeClassEqs[Reader[_]](reader) =>
         request: Request =>
           val reader = requestCallback(request).asInstanceOf[Reader[_]]
-          val streamingResponse = StreamingResponse.apply(
+          val depStreamingResponse = DeprecatedStreamingResponse.apply(
             toBuf = mapper.writeValueAsBuf,
             prefix = jsonPrefix,
             separator = jsonSeparator,
             suffix = jsonSuffix)(asyncStream = Reader.toAsyncStream(reader))
-          streamingResponse.toFutureFinagleResponse
-      case streamingResponse if runtimeClassEqs[StreamingResponse[_, _]](streamingResponse) =>
+          depStreamingResponse.toFutureFinagleResponse
+      case depStreamingResponse
+          if runtimeClassEqs[DeprecatedStreamingResponse[_, _]](depStreamingResponse) =>
         request: Request =>
-          requestCallback(request).asInstanceOf[StreamingResponse[_, _]].toFutureFinagleResponse
+          requestCallback(request)
+            .asInstanceOf[DeprecatedStreamingResponse[_, _]].toFutureFinagleResponse
+      case streamingResponse if runtimeClassEqs[StreamingResponse[Any, _]](streamingResponse) =>
+        request: Request =>
+          streamingResponse.typeArguments.head match {
+            case r: Reader[_] =>
+              requestCallback(request).asInstanceOf[StreamingResponse[Reader, _]].toFutureResponse
+            case as: AsyncStream[_] =>
+              requestCallback(request)
+                .asInstanceOf[StreamingResponse[AsyncStream, _]].toFutureResponse
+            case _ =>
+              requestCallback(request).asInstanceOf[StreamingResponse[Any, _]].toFutureResponse
+          }
     }
   }
 

http/src/main/scala/com/twitter/finatra/http/streaming/StreamingRequest.scala

@@ -4,6 +4,7 @@ import com.twitter.concurrent.AsyncStream
 import com.twitter.finagle.http.Request
 import com.twitter.finatra.json.internal.streaming.JsonStreamParser
 import com.twitter.io.{Buf, Reader}
+import scala.language.higherKinds
 
 private[http] object StreamingRequest {
 

http/src/main/scala/com/twitter/finatra/http/streaming/StreamingResponse.scala

@@ -2,7 +2,9 @@ package com.twitter.finatra.http.streaming
 
 import com.twitter.finagle.http.{Response, Status, Version}
 import com.twitter.finatra.json.FinatraObjectMapper
+import com.twitter.io.Buf
 import com.twitter.util.Future
+import scala.language.higherKinds
 
 /**
  * StreamingResponse is an abstraction over an output Primitive Stream - Reader or AsyncStream.
@@ -16,8 +18,9 @@ private[http] final case class StreamingResponse[F[_]: ToReader, A] private (
   mapper: FinatraObjectMapper,
   val stream: F[A]) {
 
-  private[this] val reader = implicitly[ToReader[F]].apply(stream).map { i =>
-    mapper.writeValueAsBuf(i)
+  private[this] val reader = implicitly[ToReader[F]].apply(stream).map {
+    case str: String => Buf.Utf8(str)
+    case any => mapper.writeValueAsBuf(any)
   }
 
   /**

http/src/test/java/com/twitter/finatra/http/tests/streaming/StreamingResponseJavaTest.java

@@ -81,7 +81,7 @@ public void serdeReaderofString() throws Exception {
     Response response = fromReader(reader);
     Future<Buf> fBuf = Readers.readAll(response.reader());
     String result = Buf.decodeString(Await.result(fBuf), StandardCharsets.UTF_8);
-    Assert.assertEquals("\"first\"\"second\"\"third\"", result);
+    Assert.assertEquals("firstsecondthird", result);
   }
 
   @Test
@@ -122,7 +122,7 @@ public void serdeAsyncStreamOfString() throws Exception {
     Response response = fromStream(stream);
     Future<Buf> fBuf = Readers.readAll(response.reader());
     String result = Buf.decodeString(Await.result(fBuf), StandardCharsets.UTF_8);
-    Assert.assertEquals("\"first\"\"second\"\"third\"", result);
+    Assert.assertEquals("firstsecondthird", result);
   }
 
 }

http/src/test/scala/com/twitter/finatra/http/EmbeddedHttpServer.scala

@@ -51,7 +51,7 @@ import scala.collection.JavaConverters._
  *                              underlying server when testing with an injectable server. By default
  *                              an injectable server under test will have an [[com.twitter.finagle.stats.InMemoryStatsReceiver]]
  *                              implementation bound for the purpose of testing. In some cases, users may want to test using
- *                              a custom [[StatsReceiver]] implementation instead and can provide and instance
+ *                              a custom [[StatsReceiver]] implementation instead and can provide an instance
  *                              to use here. For non-injectable servers this can be a shared reference
  *                              used in the server under test.
  */

http/src/test/scala/com/twitter/finatra/http/tests/integration/tweetexample/main/controllers/TweetsController.scala

@@ -3,10 +3,13 @@ package com.twitter.finatra.http.tests.integration.tweetexample.main.controllers
 import com.twitter.concurrent.AsyncStream
 import com.twitter.finagle.http.{Fields, Request, Response, Status}
 import com.twitter.finatra.http.Controller
+import com.twitter.finatra.http.response.{
+  StreamingResponseUtils,
+  StreamingResponse => DeprecatedStreamingResponse
+}
+import com.twitter.finatra.http.streaming.StreamingRequest
 import com.twitter.finatra.http.tests.integration.tweetexample.main.domain.Tweet
 import com.twitter.finatra.http.tests.integration.tweetexample.main.services.TweetsRepository
-import com.twitter.finatra.http.response.{StreamingResponse, StreamingResponseUtils}
-import com.twitter.finatra.http.streaming.StreamingRequest
 import com.twitter.io.{Buf, Reader}
 import com.twitter.util.{Duration, Future, Try}
 import java.nio.charset.StandardCharsets
@@ -16,8 +19,7 @@ import scala.collection.mutable
 class TweetsController @Inject()(
   tweetsRepository: TweetsRepository,
   onWriteLog: mutable.ArrayBuffer[String]
-)
-  extends Controller {
+) extends Controller {
 
   get("/tweets/hello") { request: Request =>
     "hello world"
@@ -47,17 +49,24 @@ class TweetsController @Inject()(
     bufs
   }
 
-
   get("/tweets/streaming_json") { request: Request =>
     tweetsRepository.getByIds(Reader.fromSeq(Seq(0, 1, 2, 3, 4, 5)))
   }
 
   get("/tweets/streaming_custom_tobuf") { request: Request =>
-    StreamingResponse(Buf.Utf8.apply) {
+    DeprecatedStreamingResponse(Buf.Utf8.apply) {
       AsyncStream("A", "B", "C")
     }
   }
 
+  get("/tweets/streamingRep_with_asyncStream") { _: Request =>
+    response.streaming(AsyncStream("A", "B", "C"))
+  }
+
+  get("/tweets/streamingRep_with_reader") { _: Request =>
+    response.streaming(Reader.fromSeq(Seq(1, 2, 3)))
+  }
+
   get("/tweets/streaming_with_transformer") { _: Request =>
     def lowercaseTransformer(as: AsyncStream[String]) = as.map(_.toLowerCase)
 
@@ -67,20 +76,27 @@ class TweetsController @Inject()(
 
     def onWrite(ignored: Unit, buf: Buf)(t: Try[Unit]): Unit = ()
 
-    StreamingResponse(
-        transformer,
-        Status.Ok,
-        Map.empty,
-        onWrite,
-        () => (),
-        Duration.Zero
-      ) {
+    DeprecatedStreamingResponse(
+      transformer,
+      Status.Ok,
+      Map.empty,
+      onWrite,
+      () => (),
+      Duration.Zero
+    ) {
       AsyncStream("A", "B", "C")
     }
   }
 
-  get("/tweets/streaming_with_onWrite") { _: Request =>
+  get("/tweets/streamingRep_with_transformer_asyncStream") { _: Request =>
+    response.streaming(AsyncStream("A", "B", "C").map(_.toLowerCase))
+  }
+
+  get("/tweets/streamingRep_with_transformer_reader") { _: Request =>
+    response.streaming(Reader.fromSeq(Seq("A", "B", "C")).map(_.toLowerCase))
+  }
 
+  get("/tweets/streaming_with_onWrite") { _: Request =>
     def serializeAndLowercase(as: AsyncStream[String]): AsyncStream[(String, Buf)] = {
       as.map(str => (str.toLowerCase, Buf.Utf8(str)))
     }
@@ -89,7 +105,7 @@ class TweetsController @Inject()(
       onWriteLog.append(lowerCased)
     }
 
-    StreamingResponse(
+    DeprecatedStreamingResponse(
       serializeAndLowercase,
       Status.Ok,
       Map.empty,
@@ -108,7 +124,7 @@ class TweetsController @Inject()(
       Fields.Pragma -> "no-cache"
     )
 
-    StreamingResponse(Buf.Utf8.apply, Status.Created, headers) {
+    DeprecatedStreamingResponse(Buf.Utf8.apply, Status.Created, headers) {
       AsyncStream("A", "B", "C")
     }
   }

http/src/test/scala/com/twitter/finatra/http/tests/integration/tweetexample/test/TweetsControllerIntegrationTest.scala

@@ -197,10 +197,32 @@ class TweetsControllerIntegrationTest extends FeatureTest {
     server.httpGet("/tweets/streaming_custom_tobuf", andExpect = Status.Ok, withBody = "ABC")
   }
 
+  test("get StreamingResponse with AsyncStream") {
+    server.httpGet("/tweets/streamingRep_with_asyncStream", andExpect = Status.Ok, withBody = "ABC")
+  }
+
+  test("get StreamingResponse with Reader") {
+    server.httpGet("/tweets/streamingRep_with_reader", andExpect = Status.Ok, withBody = "123")
+  }
+
   test("get streaming with transformer") {
     server.httpGet("/tweets/streaming_with_transformer", andExpect = Status.Ok, withBody = "abc")
   }
 
+  test("get StreamingResponse with AsyncStream with transformer ") {
+    server.httpGet(
+      "/tweets/streamingRep_with_transformer_asyncStream",
+      andExpect = Status.Ok,
+      withBody = "abc")
+  }
+
+  test("get StreamingResponse with Reader with transformer") {
+    server.httpGet(
+      "/tweets/streamingRep_with_transformer_reader",
+      andExpect = Status.Ok,
+      withBody = "abc")
+  }
+
   test("get streaming with onWrite") {
     server.httpGet("/tweets/streaming_with_onWrite", andExpect = Status.Ok, withBody = "ABC")
     assert(onWriteLog == Seq("a", "b", "c"))