core: remove complex graph building from streamed entity rendering

It turned out that aside from materialization which is still somewhat slow
at least some bits of slowness were transferred from materialization to building
stream graphs. The culprit in this case is `Source.concat`, which is really
slow.

akka-http-core/src/main/mima-filters/10.1.9.backwards.excludes/response-rendering-performance-improvements.excludes

@@ -0,0 +1,2 @@
+# Changes to internal classes
+ProblemFilters.exclude[Problem]("akka.http.impl.engine.rendering.*")

akka-http-core/src/main/scala/akka/http/impl/engine/rendering/HttpResponseRendererFactory.scala

@@ -21,6 +21,7 @@ import RenderSupport._
 import HttpProtocols._
 import akka.annotation.InternalApi
 import ResponseRenderingContext.CloseRequested
+import akka.stream.scaladsl.Sink
 import headers._
 
 import scala.util.control.NonFatal
@@ -81,9 +82,11 @@ private[http] class HttpResponseRendererFactory(
               case Strict(outElement) =>
                 push(out, outElement)
                 if (close) completeStage()
-              case Streamed(outStream) =>
-                try transfer(outStream)
-                catch {
+              case HeadersAndStreamedEntity(headerData, outStream) =>
+                try {
+                  push(out, ResponseRenderingOutput.HttpData(headerData))
+                  transfer(outStream)
+                } catch {
                   case NonFatal(e) =>
                     transferring = false
                     log.error(e, s"Rendering of response failed because response entity stream materialization failed with '${e.getMessage}'. Sending out 500 response instead.")
@@ -99,9 +102,9 @@ private[http] class HttpResponseRendererFactory(
           def onPull(): Unit = if (!hasBeenPulled(in)) tryPull(in)
         }
         setHandler(out, waitForDemandHandler)
-        def transfer(outStream: Source[ResponseRenderingOutput, Any]): Unit = {
+        def transfer(outStream: Source[ByteString, Any]): Unit = {
           transferring = true
-          val sinkIn = new SubSinkInlet[ResponseRenderingOutput]("RenderingSink")
+          val sinkIn = new SubSinkInlet[ByteString]("RenderingSink")
           def stopTransfer(): Unit = {
             transferring = false
             setHandler(out, waitForDemandHandler)
@@ -110,7 +113,7 @@ private[http] class HttpResponseRendererFactory(
           }
 
           sinkIn.setHandler(new InHandler {
-            override def onPush(): Unit = push(out, sinkIn.grab())
+            override def onPush(): Unit = push(out, ResponseRenderingOutput.HttpData(sinkIn.grab()))
             override def onUpstreamFinish(): Unit =
               if (close) completeStage()
               else stopTransfer()
@@ -125,7 +128,7 @@ private[http] class HttpResponseRendererFactory(
 
           try {
             outStream.runWith(sinkIn.sink)(interpreter.subFusingMaterializer)
-            sinkIn.pull()
+            if (isAvailable(out)) sinkIn.pull()
           } catch {
             case NonFatal(e) =>
               stopTransfer()
@@ -240,8 +243,16 @@ private[http] class HttpResponseRendererFactory(
           def renderContentLengthHeader(contentLength: Long) =
             if (status.allowsEntity) r ~~ `Content-Length` ~~ contentLength ~~ CrLf else r
 
-          def byteStrings(entityBytes: => Source[ByteString, Any]): Source[ResponseRenderingOutput, Any] =
-            renderByteStrings(r.asByteString, entityBytes, skipEntity = noEntity).map(ResponseRenderingOutput.HttpData(_))
+          def headersAndEntity(entityBytes: => Source[ByteString, Any]): StrictOrStreamed =
+            if (noEntity) {
+              entityBytes.runWith(Sink.cancelled)(subFusingMaterializer)
+              Strict(ResponseRenderingOutput.HttpData(r.asByteString))
+            } else {
+              HeadersAndStreamedEntity(
+                r.asByteString,
+                entityBytes
+              )
+            }
 
           @tailrec def completeResponseRendering(entity: ResponseEntity): StrictOrStreamed =
             entity match {
@@ -269,20 +280,20 @@ private[http] class HttpResponseRendererFactory(
                 renderHeaders(headers)
                 renderEntityContentType(r, entity)
                 renderContentLengthHeader(contentLength) ~~ CrLf
-                Streamed(byteStrings(data.via(CheckContentLengthTransformer.flow(contentLength))))
+                headersAndEntity(data.via(CheckContentLengthTransformer.flow(contentLength)))
 
               case HttpEntity.CloseDelimited(_, data) =>
                 renderHeaders(headers, alwaysClose = ctx.requestMethod != HttpMethods.HEAD)
                 renderEntityContentType(r, entity) ~~ CrLf
-                Streamed(byteStrings(data))
+                headersAndEntity(data)
 
               case HttpEntity.Chunked(contentType, chunks) =>
                 if (ctx.requestProtocol == `HTTP/1.0`)
                   completeResponseRendering(HttpEntity.CloseDelimited(contentType, chunks.map(_.data)))
                 else {
                   renderHeaders(headers)
                   renderEntityContentType(r, entity) ~~ CrLf
-                  Streamed(byteStrings(chunks.via(ChunkTransformer.flow)))
+                  headersAndEntity(chunks.via(ChunkTransformer.flow))
                 }
             }
 
@@ -293,7 +304,7 @@ private[http] class HttpResponseRendererFactory(
 
     sealed trait StrictOrStreamed
     case class Strict(bytes: ResponseRenderingOutput) extends StrictOrStreamed
-    case class Streamed(source: Source[ResponseRenderingOutput, Any]) extends StrictOrStreamed
+    case class HeadersAndStreamedEntity(headerBytes: ByteString, remainingData: Source[ByteString, Any]) extends StrictOrStreamed
   }
 
   sealed trait CloseMode