Repository: http4s/http4s
Pull request #4412 (2021-02-13 23:22:43)

S22 to main


Patch:
diff --git a/blaze-server/src/main/scala/org/http4s/server/blaze/BlazeServerBuilder.scala b/blaze-server/src/main/scala/org/http4s/server/blaze/BlazeServerBuilder.scala
index f747c80edf..4b8a1c56f6 100644
--- a/blaze-server/src/main/scala/org/http4s/server/blaze/BlazeServerBuilder.scala
+++ b/blaze-server/src/main/scala/org/http4s/server/blaze/BlazeServerBuilder.scala
@@ -45,6 +45,7 @@ import org.http4s.blaze.pipeline.stages.SSLStage
 import org.http4s.blaze.util.TickWheelExecutor
 import org.http4s.blazecore.{BlazeBackendBuilder, tickWheelResource}
 import org.http4s.internal.threads.threadFactory
+import org.http4s.internal.tls.{deduceKeyLength, getCertChain}
 import org.http4s.server.ServerRequestKeys
 import org.http4s.server.SSLKeyStoreSupport.StoreInfo
 import org.http4s.server.blaze.BlazeServerBuilder._
@@ -283,8 +284,8 @@ class BlazeServerBuilder[F[_]] private (
                     (
                       Option(session.getId).map(ByteVector(_).toHex),
                       Option(session.getCipherSuite),
-                      Option(session.getCipherSuite).map(SSLContextFactory.deduceKeyLength),
-                      SSLContextFactory.getCertChain(session).some).mapN(SecureSession.apply)
+                      Option(session.getCipherSuite).map(deduceKeyLength),
+                      getCertChain(session).some).mapN(SecureSession.apply)
                   }
               )
         case _ =>
diff --git a/blaze-server/src/main/scala/org/http4s/server/blaze/SSLContextFactory.scala b/blaze-server/src/main/scala/org/http4s/server/blaze/SSLContextFactory.scala
index 55f72c4ed1..363e0f3888 100644
--- a/blaze-server/src/main/scala/org/http4s/server/blaze/SSLContextFactory.scala
+++ b/blaze-server/src/main/scala/org/http4s/server/blaze/SSLContextFactory.scala
@@ -16,64 +16,14 @@
 
 package org.http4s.server.blaze
 
-import java.io.ByteArrayInputStream
-import java.security.cert.{CertificateFactory, X509Certificate}
-
+import java.security.cert.X509Certificate
 import javax.net.ssl.SSLSession
 
-import scala.util.Try
-
-/** Based on SSLContextFactory from jetty.
-  */
+@deprecated("Moved to org.http4s.internal.tls", "0.21.19")
 private[blaze] object SSLContextFactory {
-
-  /** Return X509 certificates for the session.
-    *
-    * @param sslSession Session from which certificate to be read
-    * @return Empty array if no certificates can be read from {{{sslSession}}}
-    */
   def getCertChain(sslSession: SSLSession): List[X509Certificate] =
-    Try {
-      val cf = CertificateFactory.getInstance("X.509")
-      sslSession.getPeerCertificates.map { certificate =>
-        val stream = new ByteArrayInputStream(certificate.getEncoded)
-        cf.generateCertificate(stream).asInstanceOf[X509Certificate]
-      }
-    }.toOption.getOrElse(Array.empty[X509Certificate]).toList
+    org.http4s.internal.tls.getCertChain(sslSession)
 
-  /** Given the name of a TLS/SSL cipher suite, return an int representing it effective stream
-    * cipher key strength. i.e. How much entropy material is in the key material being fed into the
-    * encryption routines.
-    *
-    * This is based on the information on effective key lengths in RFC 2246 - The TLS Protocol
-    * Version 1.0, Appendix C. CipherSuite definitions:
-    * <pre>
-    *                         Effective
-    *     Cipher       Type    Key Bits
-    *
-    *     NULL       * Stream     0
-    *     IDEA_CBC     Block    128
-    *     RC2_CBC_40 * Block     40
-    *     RC4_40     * Stream    40
-    *     RC4_128      Stream   128
-    *     DES40_CBC  * Block     40
-    *     DES_CBC      Block     56
-    *     3DES_EDE_CBC Block    168
-    * </pre>
-    *
-    * @param cipherSuite String name of the TLS cipher suite.
-    * @return int indicating the effective key entropy bit-length.
-    */
   def deduceKeyLength(cipherSuite: String): Int =
-    if (cipherSuite == null) 0
-    else if (cipherSuite.contains("WITH_AES_256_")) 256
-    else if (cipherSuite.contains("WITH_RC4_128_")) 128
-    else if (cipherSuite.contains("WITH_AES_128_")) 128
-    else if (cipherSuite.contains("WITH_RC4_40_")) 40
-    else if (cipherSuite.contains("WITH_3DES_EDE_CBC_")) 168
-    else if (cipherSuite.contains("WITH_IDEA_CBC_")) 128
-    else if (cipherSuite.contains("WITH_RC2_CBC_40_")) 40
-    else if (cipherSuite.contains("WITH_DES40_CBC_")) 40
-    else if (cipherSuite.contains("WITH_DES_CBC_")) 56
-    else 0
+    org.http4s.internal.tls.deduceKeyLength(cipherSuite)
 }
diff --git a/core/src/main/scala/org/http4s/internal/tls.scala b/core/src/main/scala/org/http4s/internal/tls.scala
new file mode 100644
index 0000000000..3c9551ee47
--- /dev/null
+++ b/core/src/main/scala/org/http4s/internal/tls.scala
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2013 http4s.org
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.http4s.internal
+
+import java.io.ByteArrayInputStream
+import java.security.cert.{CertificateFactory, X509Certificate}
+
+import javax.net.ssl.SSLSession
+
+import scala.util.Try
+
+/** Based on SSLContextFactory from jetty. */
+object tls {
+
+  /** Return X509 certificates for the session.
+    *
+    * @param sslSession Session from which certificate to be read
+    * @return Empty array if no certificates can be read from {{{sslSession}}}
+    */
+  def getCertChain(sslSession: SSLSession): List[X509Certificate] =
+    Try {
+      val cf = CertificateFactory.getInstance("X.509")
+      sslSession.getPeerCertificates.map { certificate =>
+        val stream = new ByteArrayInputStream(certificate.getEncoded)
+        cf.generateCertificate(stream).asInstanceOf[X509Certificate]
+      }
+    }.toOption.getOrElse(Array.empty[X509Certificate]).toList
+
+  /** Given the name of a TLS/SSL cipher suite, return an int representing it effective stream
+    * cipher key strength. i.e. How much entropy material is in the key material being fed into the
+    * encryption routines.
+    *
+    * This is based on the information on effective key lengths in RFC 2246 - The TLS Protocol
+    * Version 1.0, Appendix C. CipherSuite definitions:
+    * <pre>
+    *                         Effective
+    *     Cipher       Type    Key Bits
+    *
+    *     NULL       * Stream     0
+    *     IDEA_CBC     Block    128
+    *     RC2_CBC_40 * Block     40
+    *     RC4_40     * Stream    40
+    *     RC4_128      Stream   128
+    *     DES40_CBC  * Block     40
+    *     DES_CBC      Block     56
+    *     3DES_EDE_CBC Block    168
+    * </pre>
+    *
+    * @param cipherSuite String name of the TLS cipher suite.
+    * @return int indicating the effective key entropy bit-length.
+    */
+  def deduceKeyLength(cipherSuite: String): Int =
+    if (cipherSuite == null) 0
+    else if (cipherSuite.contains("WITH_AES_256_")) 256
+    else if (cipherSuite.contains("WITH_RC4_128_")) 128
+    else if (cipherSuite.contains("WITH_AES_128_")) 128
+    else if (cipherSuite.contains("WITH_RC4_40_")) 40
+    else if (cipherSuite.contains("WITH_3DES_EDE_CBC_")) 168
+    else if (cipherSuite.contains("WITH_IDEA_CBC_")) 128
+    else if (cipherSuite.contains("WITH_RC2_CBC_40_")) 40
+    else if (cipherSuite.contains("WITH_DES40_CBC_")) 40
+    else if (cipherSuite.contains("WITH_DES_CBC_")) 56
+    else 0
+}
diff --git a/ember-client/src/main/scala/org/http4s/ember/client/EmberClient.scala b/ember-client/src/main/scala/org/http4s/ember/client/EmberClient.scala
index 3e29413930..602847feac 100644
--- a/ember-client/src/main/scala/org/http4s/ember/client/EmberClient.scala
+++ b/ember-client/src/main/scala/org/http4s/ember/client/EmberClient.scala
@@ -23,7 +23,7 @@ import org.typelevel.keypool._
 
 final class EmberClient[F[_]] private[client] (
     private val client: Client[F],
-    private val pool: KeyPool[F, RequestKey, (RequestKeySocket[F], F[Unit])]
+    private val pool: KeyPool[F, RequestKey, EmberConnection[F]]
 )(implicit F: MonadCancelThrow[F])
     extends DefaultClient[F] {
 
diff --git a/ember-client/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala b/ember-client/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala
index bef0aa2d86..ebe6e1f282 100644
--- a/ember-client/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala
+++ b/ember-client/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala
@@ -19,6 +19,7 @@ package org.http4s.ember.client
 import cats._
 import cats.syntax.all._
 import cats.effect._
+
 import scala.concurrent.duration._
 import org.http4s.ProductId
 import org.http4s.client._
@@ -28,8 +29,9 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
 import fs2.io.tcp.SocketGroup
 import fs2.io.tcp.SocketOptionMapping
 import fs2.io.tls._
+
 import scala.concurrent.duration.Duration
-import org.http4s.headers.{Connection, `User-Agent`}
+import org.http4s.headers.{`User-Agent`}
 import org.http4s.ember.client.internal.ClientHelpers
 
 final class EmberClientBuilder[F[_]: Async] private (
@@ -111,25 +113,20 @@ final class EmberClientBuilder[F[_]: Async] private (
       )
       builder =
         KeyPoolBuilder
-          .apply[F, RequestKey, (RequestKeySocket[F], F[Unit])](
-            (requestKey: RequestKey) => {
-              val alloced =
+          .apply[F, RequestKey, EmberConnection[F]](
+            (requestKey: RequestKey) =>
+              EmberConnection(
                 org.http4s.ember.client.internal.ClientHelpers
                   .requestKeyToSocketWithKey[F](
                     requestKey,
                     tlsContextOptWithDefault,
                     sg,
                     additionalSocketOptions
-                  )
-                  .allocated
-              alloced <* logger.trace(s"Created Connection - RequestKey: ${requestKey}")
-            },
-            { case (RequestKeySocket(socket, r), shutdown) =>
-              logger.trace(s"Shutting Down Connection - RequestKey: ${r}") >>
-                socket.endOfInput.attempt.void >>
-                socket.endOfOutput.attempt.void >>
-                socket.close.attempt.void >>
-                shutdown.attempt.void
+                  )) <* logger.trace(s"Created Connection - RequestKey: ${requestKey}"),
+            { case connection =>
+              logger.trace(
+                s"Shutting Down Connection - RequestKey: ${connection.keySocket.requestKey}") >>
+                connection.cleanup
             }
           )
           .withDefaultReuseState(Reusable.DontReuse)
@@ -145,38 +142,34 @@ final class EmberClientBuilder[F[_]: Async] private (
           _ <- Resource.eval(
             pool.state.flatMap { poolState =>
               logger.trace(
-                s"Connection Taken - Key: ${managed.value._1.requestKey} - Reused: ${managed.isReused} - PoolState: $poolState"
+                s"Connection Taken - Key: ${managed.value.keySocket.requestKey} - Reused: ${managed.isReused} - PoolState: $poolState"
               )
             }
           )
-          responseResource <-
-            Resource.eval(
-              ClientHelpers
-                .request[F](
+          responseResource <- Resource.makeCase(
+            ClientHelpers
+              .request[F](
+                request,
+                managed.value,
+                chunkSize,
+                maxResponseHeaderSize,
+                idleConnectionTime,
+                timeout,
+                userAgent
+              )
+          ) { case ((response, drain), exitCase) =>
+            exitCase match {
+              case Resource.ExitCase.Succeeded =>
+                ClientHelpers.postProcessResponse(
                   request,
-                  managed.value._1,
-                  managed.canBeReused,
-                  chunkSize,
-                  maxResponseHeaderSize,
-                  idleConnectionTime,
-                  timeout,
-                  userAgent
-                )
-                .map(response =>
-                  // TODO If Response Body has a take(1).compile.drain - would leave rest of bytes in root stream for next caller
-                  response.copy(body = response.body.onFinalizeCaseWeak {
-                    case Resource.ExitCase.Succeeded =>
-                      val requestClose = request.headers.get(Connection).exists(_.hasClose)
-                      val responseClose = response.isChunked || response.headers
-                        .get(Connection)
-                        .exists(_.hasClose)
-
-                      if (requestClose || responseClose) Sync[F].unit
-                      else managed.canBeReused.set(Reusable.Reuse)
-                    case Resource.ExitCase.Canceled => Sync[F].unit
-                    case Resource.ExitCase.Errored(_) => Sync[F].unit
-                  })))
-        } yield responseResource
+                  response,
+                  drain,
+                  managed.value.nextBytes,
+                  managed.canBeReused)
+              case _ => Applicative[F].unit
+            }
+          }
+        } yield responseResource._1
       }
       new EmberClient[F](client, pool)
     }
diff --git a/ember-client/src/main/scala/org/http4s/ember/client/EmberConnection.scala b/ember-client/src/main/scala/org/http4s/ember/client/EmberConnection.scala
new file mode 100644
index 0000000000..2e471ec3bb
--- /dev/null
+++ b/ember-client/src/main/scala/org/http4s/ember/client/EmberConnection.scala
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2019 http4s.org
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.http4s.ember.client
+
+import cats._
+import cats.effect.{Concurrent, Resource}
+import cats.syntax.all._
+import cats.effect.kernel.Ref
+
+private[ember] final case class EmberConnection[F[_]](
+    keySocket: RequestKeySocket[F],
+    shutdown: F[Unit],
+    nextBytes: Ref[F, Array[Byte]])(implicit F: MonadThrow[F]) {
+  def cleanup: F[Unit] =
+    nextBytes.set(Array.emptyByteArray) >>
+      keySocket.socket.endOfInput.attempt.void >>
+      keySocket.socket.endOfOutput.attempt.void >>
+      keySocket.socket.close.attempt.void >>
+      shutdown.attempt.void
+}
+
+private[ember] object EmberConnection {
+  def apply[F[_]: Concurrent](
+      keySocketResource: Resource[F, RequestKeySocket[F]]): F[EmberConnection[F]] =
+    Ref[F].of(Array.emptyByteArray).flatMap { nextBytes =>
+      keySocketResource.allocated.map { case (keySocket, release) =>
+        EmberConnection(keySocket, release, nextBytes)
+      }
+    }
+}
diff --git a/ember-client/src/main/scala/org/http4s/ember/client/internal/ClientHelpers.scala b/ember-client/src/main/scala/org/http4s/ember/client/internal/ClientHelpers.scala
index e2e097c12d..046bc7bb89 100644
--- a/ember-client/src/main/scala/org/http4s/ember/client/internal/ClientHelpers.scala
+++ b/ember-client/src/main/scala/org/http4s/ember/client/internal/ClientHelpers.scala
@@ -81,14 +81,13 @@ private[client] object ClientHelpers {
 
   def request[F[_]: Async](
       request: Request[F],
-      requestKeySocket: RequestKeySocket[F],
-      reuseable: Ref[F, Reusable],
+      connection: EmberConnection[F],
       chunkSize: Int,
       maxResponseHeaderSize: Int,
       idleTimeout: Duration,
       timeout: Duration,
       userAgent: Option[`User-Agent`]
-  ): F[Response[F]] = {
+  ): F[(Response[F], F[Option[Array[Byte]]])] = {
 
     def writeRequestToSocket(
         req: Request[F],
@@ -100,19 +99,20 @@ private[client] object ClientHelpers {
         .compile
         .drain
 
-    def writeRead(req: Request[F]): F[Response[F]] =
-      writeRequestToSocket(req, requestKeySocket.socket, durationToFinite(idleTimeout)) >> {
-        Parser.Response
-          .parser(maxResponseHeaderSize, durationToFinite(timeout))(
-            requestKeySocket.socket.reads(chunkSize, durationToFinite(idleTimeout))
-          )
-          .map(_._1)
-      }
+    def writeRead(req: Request[F]): F[(Response[F], F[Option[Array[Byte]]])] =
+      writeRequestToSocket(req, connection.keySocket.socket, durationToFinite(idleTimeout)) >>
+        connection.nextBytes.getAndSet(Array.emptyByteArray).flatMap { head =>
+          Parser.Response
+            .parser(maxResponseHeaderSize, durationToFinite(timeout))(
+              head,
+              connection.keySocket.socket.read(chunkSize, durationToFinite(idleTimeout))
+            )
+        }
 
     for {
       processedReq <- preprocessRequest(request, userAgent)
-      resp <- writeRead(processedReq)
-    } yield postProcessResponse(processedReq, resp, reuseable)
+      res <- writeRead(processedReq)
+    } yield res
   }
 
   private[internal] def preprocessRequest[F[_]: Monad: Clock](
@@ -129,24 +129,21 @@ private[client] object ClientHelpers {
       .putHeaders(userAgentHeader.toSeq: _*)
   }
 
-  private[internal] def postProcessResponse[F[_]: Concurrent](
+  private[ember] def postProcessResponse[F[_]](
       req: Request[F],
       resp: Response[F],
-      canBeReused: Ref[F, Reusable]): Response[F] = {
-    val out = resp.copy(
-      body = resp.body.onFinalizeCaseWeak {
-        case Resource.ExitCase.Succeeded =>
-          val requestClose = req.headers.get(Connection).exists(_.hasClose)
-          val responseClose = resp.headers.get(Connection).exists(_.hasClose)
+      drain: F[Option[Array[Byte]]],
+      nextBytes: Ref[F, Array[Byte]],
+      canBeReused: Ref[F, Reusable])(implicit F: Concurrent[F]): F[Unit] =
+    drain.flatMap {
+      case Some(bytes) =>
+        val requestClose = req.headers.get(Connection).exists(_.hasClose)
+        val responseClose = resp.headers.get(Connection).exists(_.hasClose)
 
-          if (requestClose || responseClose) Applicative[F].unit
-          else canBeReused.set(Reusable.Reuse)
-        case Resource.ExitCase.Canceled => Applicative[F].unit
-        case Resource.ExitCase.Errored(_) => Applicative[F].unit
-      }
-    )
-    out
-  }
+        if (requestClose || responseClose) F.unit
+        else nextBytes.set(bytes) >> canBeReused.set(Reusable.Reuse)
+      case None => F.unit
+    }
 
   // https://github.com/http4s/http4s/blob/main/blaze-client/src/main/scala/org/http4s/client/blaze/Http1Support.scala#L86
   private def getAddress[F[_]: Sync](requestKey: RequestKey): F[InetSocketAddress] =
@@ -159,11 +156,11 @@ private[client] object ClientHelpers {
 
   // Assumes that the request doesn't have fancy finalizers besides shutting down the pool
   private[client] def getValidManaged[F[_]: Sync](
-      pool: KeyPool[F, RequestKey, (RequestKeySocket[F], F[Unit])],
-      request: Request[F]): Resource[F, Managed[F, (RequestKeySocket[F], F[Unit])]] =
+      pool: KeyPool[F, RequestKey, EmberConnection[F]],
+      request: Request[F]): Resource[F, Managed[F, EmberConnection[F]]] =
     pool.take(RequestKey.fromRequest(request)).flatMap { managed =>
       Resource
-        .eval(managed.value._1.socket.isOpen)
+        .eval(managed.value.keySocket.socket.isOpen)
         .ifM(
           managed.pure[Resource[F, *]],
           // Already Closed,
diff --git a/ember-core/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala b/ember-core/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala
index f4ea245238..f2311e7487 100644
--- a/ember-core/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala
+++ b/ember-core/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala
@@ -12,7 +12,7 @@ package org.http4s.ember.core
 
 import cats._
 import cats.syntax.all._
-import cats.effect.kernel.Deferred
+import cats.effect.kernel.{Deferred, Ref}
 import fs2._
 import scodec.bits.ByteVector
 import Shared._
@@ -26,14 +26,21 @@ private[ember] object ChunkedEncoding {
     * decodes from the HTTP chunked encoding. After last chunk this terminates. Allows to specify max header size, after which this terminates
     * Please see https://en.wikipedia.org/wiki/Chunked_transfer_encoding for details
     */
-  def decode[F[_]](maxChunkHeaderSize: Int, trailers: Deferred[F, Headers])(implicit
-      F: MonadThrow[F]): Pipe[F, Byte, Byte] = {
+  def decode[F[_]](
+      head: Array[Byte],
+      read: F[Option[Chunk[Byte]]],
+      maxChunkHeaderSize: Int,
+      trailers: Deferred[F, Headers],
+      rest: Ref[F, Option[Array[Byte]]])(implicit F: MonadThrow[F]): Stream[F, Byte] = {
     // on left reading the header of chunk (acting as buffer)
     // on right reading the chunk itself, and storing remaining bytes of the chunk
-    def go(expect: Either[ByteVector, Long], in: Stream[F, Byte]): Pull[F, Byte, Unit] =
-      in.pull.uncons.flatMap {
+    def go(expect: Either[ByteVector, Long], head: Array[Byte]): Pull[F, Byte, Unit] = {
+      val uncons =
+        if (head.nonEmpty) Pull.pure(Some(Chunk.byteVector(ByteVector.apply(head))))
+        else Pull.eval(read)
+      uncons.flatMap {
         case None => Pull.done
-        case Some((h, tl)) =>
+        case Some(h) =>
           val bv = h.toByteVector
           expect match {
             case Left(header) =>
@@ -42,12 +49,12 @@ private[ember] object ChunkedEncoding {
               if (endOfheader == 0)
                 go(
                   expect,
-                  Stream.chunk(Chunk.byteVector(bv.drop(`\r\n`.size))) ++ tl
+                  bv.drop(`\r\n`.size).toArray
                 ) //strip any leading crlf on header, as this starts with /r/n
               else if (endOfheader < 0 && nh.size > maxChunkHeaderSize)
                 Pull.raiseError[F](EmberException.ChunkedEncodingError(
                   s"Failed to get Chunk header. Size exceeds max($maxChunkHeaderSize) : ${nh.size} ${nh.decodeUtf8}"))
-              else if (endOfheader < 0) go(Left(nh), tl)
+              else if (endOfheader < 0) go(Left(nh), Array.emptyByteArray)
               else {
                 val (hdr, rem) = nh.splitAt(endOfheader + `\r\n`.size)
                 readChunkedHeader(hdr.dropRight(`\r\n`.size)) match {
@@ -57,46 +64,55 @@ private[ember] object ChunkedEncoding {
                         s"Failed to parse chunked header : ${hdr.decodeUtf8}"))
                   case Some(0) =>
                     // Done With Message, Now Parse Trailers
-                    parseTrailers[F](maxChunkHeaderSize)(Stream.chunk(Chunk.byteVector(rem)) ++ tl)
-                      .flatMap { hdrs =>
-                        Pull.eval(trailers.complete(hdrs)) >> Pull.done
-                      }
-                  case Some(sz) => go(Right(sz), Stream.chunk(Chunk.byteVector(rem)) ++ tl)
+                    Pull.eval(
+                      parseTrailers[F](maxChunkHeaderSize)(rem.toArray, read)
+                        .flatMap { case (hdrs, bytes) =>
+                          trailers.complete(hdrs) >> rest.set(Some(bytes))
+                        }
+                    ) >> Pull.done
+                  case Some(sz) => go(Right(sz), rem.toArray)
                 }
               }
 
             case Right(remains) =>
               if (remains == bv.size)
-                Pull.output(Chunk.byteVector(bv)) >> go(Left(ByteVector.empty), tl)
+                Pull.output(Chunk.byteVector(bv)) >> go(
+                  Left(ByteVector.empty),
+                  Array.emptyByteArray)
               else if (remains > bv.size)
-                Pull.output(Chunk.byteVector(bv)) >> go(Right(remains - bv.size), tl)
+                Pull.output(Chunk.byteVector(bv)) >> go(
+                  Right(remains - bv.size),
+                  Array.emptyByteArray)
               else {
                 val (out, next) = bv.splitAt(remains.toLong)
-                Pull.output(Chunk.byteVector(out)) >> go(
-                  Left(ByteVector.empty),
-                  Stream.chunk(Chunk.byteVector(next)) ++ tl)
+                Pull.output(Chunk.byteVector(out)) >> go(Left(ByteVector.empty), next.toArray)
               }
           }
       }
+    }
 
-    go(Left(ByteVector.empty), _).stream
+    go(Left(ByteVector.empty), head).stream
   }
 
   private def parseTrailers[F[_]: MonadThrow](
       maxHeaderSize: Int
-  )(s: Stream[F, Byte]): Pull[F, Nothing, Headers] =
-    s.pull.uncons.flatMap {
-      case None => Pull.pure(Headers.empty)
-      case Some((chunk, tl)) =>
-        if (chunk.isEmpty) parseTrailers(maxHeaderSize)(tl)
+  )(head: Array[Byte], read: F[Option[Chunk[Byte]]]): F[(Headers, Array[Byte])] = {
+    val uncons =
+      if (head.nonEmpty) (Some(Chunk.byteVector(ByteVector(head))): Option[Chunk[Byte]]).pure[F]
+      else read
+    uncons.flatMap {
+      case None => (Headers.empty, Array.emptyByteArray).pure[F]
+      case Some(chunk) =>
+        if (chunk.isEmpty) parseTrailers(maxHeaderSize)(Array.emptyByteArray, read)
         else if (chunk.toByteVector.startsWith(Shared.`\r\n`))
-          Pull.pure(Headers.empty)
+          (Headers.empty, chunk.toArray.drop(`\r\n`.size.toInt)).pure[F]
         else
-          Parser.HeaderP.parseHeaders(Stream.chunk(chunk) ++ tl, maxHeaderSize, None).map {
-            case (headers, _, _, _) =>
-              headers
+          Parser.HeaderP.parseHeaders(chunk.toArray, read, maxHeaderSize, None).map {
+            case (headers, _, _, bytes) =>
+              (headers, bytes)
           }
     }
+  }
 
   private val lastChunk: Chunk[Byte] =
     Chunk.byteVector((ByteVector('0') ++ `\r\n` ++ `\r\n`).compact)
diff --git a/ember-core/src/main/scala/org/http4s/ember/core/Parser.scala b/ember-core/src/main/scala/org/http4s/ember/core/Parser.scala
index f3930d1c71..7b40b4cc55 100644
--- a/ember-core/src/main/scala/org/http4s/ember/core/Parser.scala
+++ b/ember-core/src/main/scala/org/http4s/ember/core/Parser.scala
@@ -18,25 +18,29 @@ package org.http4s.ember.core
 
 import cats._
 import cats.effect.{MonadThrow => _, _}
-import cats.effect.kernel.Deferred
+import cats.effect.kernel.{Deferred, Ref}
 import cats.syntax.all._
 import fs2._
 import org.http4s._
 import scala.annotation.switch
 import scala.collection.mutable
 import scala.concurrent.duration.FiniteDuration
+import scodec.bits.ByteVector
 
 private[ember] object Parser {
 
   object HeaderP {
 
-    def parseHeaders[F[_]: MonadThrow](
-        s: Stream[F, Byte],
+    def parseHeaders[F[_]](
+        head: Array[Byte],
+        read: F[Option[Chunk[Byte]]],
         maxHeaderLength: Int,
-        acc: Option[ParseHeadersIncomplete])
-        : Pull[F, Nothing, (Headers, Boolean, Option[Long], Stream[F, Byte])] =
-      s.pull.uncons.flatMap {
-        case Some((chunk, tl)) =>
+        acc: Option[ParseHeadersIncomplete])(implicit
+        F: MonadThrow[F]): F[(Headers, Boolean, Option[Long], Array[Byte])] = {
+      // TODO: improve this
+      val uncons = if (head.nonEmpty) F.pure(Some(Chunk.byteVector(ByteVector(head)))) else read
+      uncons.flatMap {
+        case Some(chunk) =>
           val nextArr: Array[Byte] = acc match {
             case None => chunk.toArray
             case Some(last) => combineArrays(last.bv, chunk.toArray)
@@ -57,12 +61,13 @@ private[ember] object Parser {
           }
           result match {
             case ParseHeadersCompleted(headers, rest, chunked, length) =>
-              Pull.pure((headers, chunked, length, Stream.chunk(Chunk.array(rest)) ++ tl))
-            case p @ ParseHeadersError(_) => Pull.raiseError[F](p)
+              F.pure((headers, chunked, length, rest))
+            case p @ ParseHeadersError(_) => F.raiseError(p)
             case p @ ParseHeadersIncomplete(_, _, _, _, _, _, _, _) =>
-              if (nextArr.size <= maxHeaderLength) parseHeaders(tl, maxHeaderLength, p.some)
+              if (nextArr.size <= maxHeaderLength)
+                parseHeaders(Array.emptyByteArray, read, maxHeaderLength, p.some)
               else
-                Pull.raiseError[F](
+                F.raiseError(
                   ParseHeadersError(
                     new Throwable(
                       s"Parse Headers Exceeded Max Content-Length current size: ${nextArr.size}, only allow ${maxHeaderLength}")
@@ -70,9 +75,10 @@ private[ember] object Parser {
                 )
           }
         case None =>
-          Pull.raiseError[F](
+          F.raiseError(
             ParseHeadersError(new Throwable("Reached Ended of Stream Looking for Headers")))
       }
+    }
 
     private val colon: Byte = ':'.toByte
     private val cr: Byte = '\r'.toByte
@@ -195,27 +201,29 @@ private[ember] object Parser {
 
       val emptyStreamError = ParsePreludeError("Cannot Parse Empty Stream", None, None, None, None)
 
-      def parsePrelude[F[_]: MonadThrow](
-          s: Stream[F, Byte],
+      def parsePrelude[F[_]](
+          head: Array[Byte],
+          read: F[Option[Chunk[Byte]]],
           maxHeaderLength: Int,
-          acc: Option[ParsePreludeIncomlete] = None)
-          : Pull[F, Nothing, (Method, Uri, HttpVersion, Stream[F, Byte])] =
-        s.pull.uncons.flatMap {
-          case Some((chunk, tl)) =>
+          acc: Option[ParsePreludeIncomplete] = None)(implicit
+          F: MonadThrow[F]): F[(Method, Uri, HttpVersion, Array[Byte])] = {
+        val uncons = if (head.nonEmpty) F.pure(Some(Chunk.byteVector(ByteVector(head)))) else read
+
+        uncons.flatMap {
+          case Some(chunk) =>
             val next: Array[Byte] = acc match {
               case None => chunk.toArray
               case Some(ic) => combineArrays(ic.bv, chunk.toArray)
             }
             ReqPrelude.preludeInSection(next) match {
               case ParsePreludeComplete(m, u, h, rest) =>
-                val out = if (rest.isEmpty) tl else Stream.chunk(Chunk.array(rest)) ++ tl
-                Pull.pure((m, u, h, out))
-              case t @ ParsePreludeError(_, _, _, _, _) => Pull.raiseError[F](t)
-              case p @ ParsePreludeIncomlete(_, _, method, uri, httpVersion) =>
+                F.pure((m, u, h, rest))
+              case t @ ParsePreludeError(_, _, _, _, _) => F.raiseError(t)
+              case p @ ParsePreludeIncomplete(_, _, method, uri, httpVersion) =>
                 if (next.size <= maxHeaderLength)
-                  parsePrelude(tl, maxHeaderLength, p.some)
+                  parsePrelude(Array.emptyByteArray, read, maxHeaderLength, p.some)
                 else
-                  Pull.raiseError[F](
+                  F.raiseError(
                     ParsePreludeError(
                       "Reached Max Header Length Looking for Request Prelude",
                       None,
@@ -225,10 +233,10 @@ private[ember] object Parser {
             }
           case None =>
             acc match {
-              case None => Pull.raiseError(emptyStreamError)
-              case Some(incomplete) if incomplete.bv.isEmpty => Pull.raiseError(emptyStreamError)
+              case None => F.raiseError(emptyStreamError)
+              case Some(incomplete) if incomplete.bv.isEmpty => F.raiseError(emptyStreamError)
               case Some(incomplete) =>
-                Pull.raiseError(
+                F.raiseError(
                   ParsePreludeError(
                     s"Unexpected EOF - $incomplete",
                     None,
@@ -237,6 +245,7 @@ private[ember] object Parser {
                     incomplete.httpVersion))
             }
         }
+      }
 
       // sealed trait ParsePreludeState
       // 0 case object GetMethod extends ParsePreludeState
@@ -258,7 +267,7 @@ private[ember] object Parser {
             caused.orNull
           )
           with ParsePreludeResult
-      final case class ParsePreludeIncomlete(
+      final case class ParsePreludeIncomplete(
           idx: Int,
           bv: Array[Byte],
           // buffer: String,
@@ -344,76 +353,85 @@ private[ember] object Parser {
         else if (method != null && uri != null && httpVersion != null)
           ParsePreludeComplete(method, uri, httpVersion, bv.drop(idx))
         else
-          ParsePreludeIncomlete(idx, bv, Option(method), Option(uri), Option(httpVersion))
+          ParsePreludeIncomplete(idx, bv, Option(method), Option(uri), Option(httpVersion))
       }
     }
 
     def parser[F[_]: Concurrent: Temporal](maxHeaderLength: Int, timeout: Option[FiniteDuration])(
-        s: Stream[F, Byte]): F[(Request[F], Stream[F, Byte])] =
+        head: Array[Byte],
+        read: F[Option[Chunk[Byte]]]
+    ): F[(Request[F], F[Option[Array[Byte]]])] =
       Deferred[F, Headers].flatMap { trailers =>
-        val baseStream = ReqPrelude
-          .parsePrelude[F](s, maxHeaderLength, None)
-          .flatMap { case (method, uri, httpVersion, rest) =>
-            HeaderP.parseHeaders(rest, maxHeaderLength, None).flatMap {
-              case (headers, chunked, contentLength, rest) =>
+        val action = ReqPrelude
+          .parsePrelude[F](head, read, maxHeaderLength, None)
+          .flatMap { case (method, uri, httpVersion, bytes) =>
+            HeaderP.parseHeaders(bytes, read, maxHeaderLength, None).flatMap {
+              case (headers, chunked, contentLength, bytes) =>
                 val baseReq: org.http4s.Request[F] = org.http4s.Request[F](
                   method = method,
                   uri = uri,
                   httpVersion = httpVersion,
                   headers = headers
                 )
-                val req: org.http4s.Request[F] =
-                  if (chunked)
-                    baseReq
-                      .withAttribute(Message.Keys.TrailerHeaders[F], trailers.get)
-                      .withBodyStream(
-                        rest.through(ChunkedEncoding.decode(maxHeaderLength, trailers)))
-                  else
-                    baseReq.withBodyStream(rest.take(contentLength.getOrElse(0L)))
-
-                Pull.output1((req, rest))
+
+                if (chunked) {
+                  Ref.of[F, Option[Array[Byte]]](None).map { rest =>
+                    (
+                      baseReq
+                        .withAttribute(Message.Keys.TrailerHeaders[F], trailers.get)
+                        .withBodyStream(
+                          ChunkedEncoding.decode(bytes, read, maxHeaderLength, trailers, rest)),
+                      rest.get)
+                  }
+                } else {
+                  Body.parseFixedBody(contentLength.getOrElse(0L), bytes, read).map {
+                    case (bodyStream, drain) =>
+                      (baseReq.withBodyStream(bodyStream), drain)
+                  }
+                }
             }
           }
-          .stream
-          .take(1)
 
-        val action = baseStream.compile.lastOrError
-        timeout match {
-          case None => action
-          case Some(timeout) => Temporal[F].timeout(action, timeout)
-        }
+        timeout.fold(action)(duration => Temporal[F].timeout(action, duration))
       }
   }
 
   object Response {
+
     def parser[F[_]: Concurrent: Temporal](maxHeaderLength: Int, timeout: Option[FiniteDuration])(
-        s: Stream[F, Byte]): F[(Response[F], Stream[F, Byte])] =
+        head: Array[Byte],
+        read: F[Option[Chunk[Byte]]]
+    ): F[(Response[F], F[Option[Array[Byte]]])] =
       Deferred[F, Headers].flatMap { trailers =>
-        val base = RespPrelude
-          .parsePrelude(s, maxHeaderLength, None)
-          .flatMap { case (httpVersion, status, s) =>
-            HeaderP.parseHeaders(s, maxHeaderLength, None).flatMap {
-              case (headers, chunked, contentLength, rest) =>
+        val action = RespPrelude
+          .parsePrelude(head, read, maxHeaderLength, None)
+          .flatMap { case (httpVersion, status, bytes) =>
+            HeaderP.parseHeaders(bytes, read, maxHeaderLength, None).flatMap {
+              case (headers, chunked, contentLength, bytes) =>
                 val baseResp = org.http4s.Response[F](
                   httpVersion = httpVersion,
                   status = status,
                   headers = headers
                 )
-                val resp: org.http4s.Response[F] =
-                  if (chunked)
-                    baseResp
-                      .withAttribute(Message.Keys.TrailerHeaders[F], trailers.get)
-                      .withBodyStream(
-                        rest.through(ChunkedEncoding.decode(maxHeaderLength, trailers)))
-                  else
-                    baseResp.withBodyStream(rest.take(contentLength.getOrElse(0L)))
-                Pull.output1((resp, rest))
+
+                if (chunked) {
+                  Ref.of[F, Option[Array[Byte]]](None).map { rest =>
+                    (
+                      baseResp
+                        .withAttribute(Message.Keys.TrailerHeaders[F], trailers.get)
+                        .withBodyStream(
+                          ChunkedEncoding.decode(bytes, read, maxHeaderLength, trailers, rest)),
+                      rest.get)
+                  }
+                } else {
+                  Body.parseFixedBody(contentLength.getOrElse(0L), bytes, read).map {
+                    case (bodyStream, drain) =>
+                      (baseResp.withBodyStream(bodyStream), drain)
+                  }
+                }
             }
           }
-          .stream
-          .take(1)
 
-        val action = base.compile.lastOrError
         timeout.fold(action)(duration => Temporal[F].timeout(action, duration))
       }
 
@@ -421,36 +439,39 @@ private[ember] object Parser {
 
       val emptyStreamError = RespPreludeError("Cannot Parse Empty Stream", None)
 
-      def parsePrelude[F[_]: MonadThrow](
-          s: Stream[F, Byte],
+      def parsePrelude[F[_]](
+          head: Array[Byte],
+          read: F[Option[Chunk[Byte]]],
           maxHeaderLength: Int,
-          acc: Option[Array[Byte]] = None)
-          : Pull[F, Nothing, (HttpVersion, Status, Stream[F, Byte])] =
-        s.pull.uncons.flatMap {
-          case Some((chunk, tl)) =>
+          acc: Option[Array[Byte]] = None)(implicit
+          F: MonadThrow[F]): F[(HttpVersion, Status, Array[Byte])] = {
+        val pull = if (head.nonEmpty) F.pure(Some(Chunk.byteVector(ByteVector(head)))) else read
+
+        pull.flatMap {
+          case Some(chunk) =>
             val next: Array[Byte] = acc match {
               case None => chunk.toArray
               case Some(remains) => combineArrays(remains, chunk.toArray)
             }
             preludeInSection(next) match {
               case RespPreludeComplete(httpVersion, status, rest) =>
-                Pull.pure((httpVersion, status, Stream.chunk(Chunk.array(rest)) ++ tl))
-              case t @ RespPreludeError(_, _) => Pull.raiseError[F](t)
+                (httpVersion, status, rest).pure[F]
+              case t @ RespPreludeError(_, _) => F.raiseError(t)
               case RespPreludeIncomplete =>
                 if (next.size <= maxHeaderLength)
-                  parsePrelude(tl, maxHeaderLength, next.some)
+                  parsePrelude(Array.emptyByteArray, read, maxHeaderLength, next.some)
                 else
-                  Pull.raiseError[F](
+                  F.raiseError(
                     RespPreludeError(
                       "Reached Max Header Length Looking for Response Prelude",
                       None))
             }
           case None =>
             acc match {
-              case None => Pull.raiseError(emptyStreamError)
-              case Some(incomplete) if incomplete.isEmpty => Pull.raiseError(emptyStreamError)
+              case None => F.raiseError(emptyStreamError)
+              case Some(incomplete) if incomplete.isEmpty => F.raiseError(emptyStreamError)
               case Some(_) =>
-                Pull.raiseError[F](
+                F.raiseError(
                   RespPreludeError(
                     "Unexpectedly Reached Ended of Stream Looking for Response Prelude",
                     None)
@@ -458,6 +479,7 @@ private[ember] object Parser {
             }
 
         }
+      }
 
       private val space = ' '.toByte
       private val cr: Byte = '\r'.toByte
@@ -540,6 +562,58 @@ private[ember] object Parser {
     }
   }
 
+  object Body {
+    def parseFixedBody[F[_]: Concurrent](
+        contentLength: Long,
+        head: Array[Byte],
+        read: F[Option[Chunk[Byte]]]): F[(EntityBody[F], F[Option[Array[Byte]]])] =
+      if (contentLength > 0) {
+        if (head.length >= contentLength) {
+          val (body, rest) = head.splitAt(contentLength.toInt)
+          (
+            Stream.chunk(Chunk.byteVector(ByteVector(body))).covary[F],
+            (Some(rest): Option[Array[Byte]]).pure[F])
+            .pure[F]
+        } else {
+          val unread = contentLength - head.length
+          Ref.of[F, Either[Long, Array[Byte]]](Left(unread)).map { state =>
+            val bodyStream = Stream.eval(state.get).flatMap {
+              case Right(_) =>
+                Stream.raiseError(new Throwable("Body has already been completely read"))
+              case Left(remaining) =>
+                readStream(read).chunks
+                  .evalMapAccumulate(remaining) { case (r, chunk) =>
+                    if (chunk.size >= r) {
+                      val (rest, after) = chunk.splitAt(r.toInt)
+                      state.set(Right(after.toArray)).as((0L, rest))
+                    } else {
+                      val r2 = r - chunk.size
+                      state.set(Left(r2)).as((r2, chunk))
+                    }
+                  }
+                  .takeThrough(_._1 > 0)
+                  .flatMap(t => Stream.chunk(t._2))
+            }
+
+            // If the remaining bytes for the body have not yet been read, close the connection.
+            // followup: Check if there are bytes immediately available without blocking
+            val drain: F[Option[Array[Byte]]] = state.get.map(_.toOption)
+
+            (Stream.chunk(Chunk.byteVector(ByteVector(head))) ++ bodyStream, drain)
+          }
+        }
+      } else {
+        (EmptyBody.covary[F], (Some(head): Option[Array[Byte]]).pure[F]).pure[F]
+      }
+
+    private def readStream[F[_]](read: F[Option[Chunk[Byte]]]): Stream[F, Byte] =
+      Stream.eval(read).flatMap {
+        case Some(bytes) =>
+          Stream.chunk(bytes) ++ readStream(read)
+        case None => Stream.empty
+      }
+  }
+
   private def combineArrays[A: scala.reflect.ClassTag](a1: Array[A], a2: Array[A]): Array[A] = {
     val buff = mutable.ArrayBuffer[A]()
     buff.++=(a1)
diff --git a/ember-server/src/main/scala/org/http4s/ember/server/internal/ServerHelpers.scala b/ember-server/src/main/scala/org/http4s/ember/server/internal/ServerHelpers.scala
index 9d21371c6f..806ce99b17 100644
--- a/ember-server/src/main/scala/org/http4s/ember/server/internal/ServerHelpers.scala
+++ b/ember-server/src/main/scala/org/http4s/ember/server/internal/ServerHelpers.scala
@@ -17,21 +17,27 @@
 package org.http4s.ember.server.internal
 
 import cats._
-import fs2._
-import fs2.io.tcp._
-import fs2.io.tls._
 import fs2.io.Network
+import cats.data.NonEmptyList
 import cats.effect._
 import cats.syntax.all._
-import scala.concurrent.duration._
+import com.comcast.ip4s.SocketAddress
+import fs2.io.tcp._
+import fs2.io.tls._
+import fs2.{Chunk, Stream}
 import java.net.InetSocketAddress
 import org.http4s._
+import org.http4s.ember.core.Util.durationToFinite
+import org.http4s.ember.core.{Encoder, Parser}
 import org.http4s.headers.{Connection, Date}
+import org.http4s.internal.tls.{deduceKeyLength, getCertChain}
+import org.http4s.server.{SecureSession, ServerRequestKeys}
 import org.typelevel.ci.CIString
-import _root_.org.http4s.ember.core.{Encoder, Parser}
-import _root_.org.typelevel.log4cats.Logger
-import cats.data.NonEmptyList
-import _root_.org.http4s.ember.core.Util.durationToFinite
+import org.typelevel.log4cats.Logger
+import org.typelevel.vault.Vault
+import scala.concurrent.duration._
+import scodec.bits.ByteVector
+
 private[server] object ServerHelpers {
 
   private val closeCi = CIString("close")
@@ -87,7 +93,8 @@ private[server] object ServerHelpers {
                 requestHeaderReceiveTimeout,
                 httpApp,
                 errorHandler,
-                onWriteFailure))
+                onWriteFailure
+              ))
       }
 
     StreamForking.forking(streams, maxConcurrency)
@@ -115,19 +122,22 @@ private[server] object ServerHelpers {
     }
 
   private[internal] def runApp[F[_]: Concurrent: Temporal](
-      incoming: Stream[F, Byte],
+      head: Array[Byte],
+      read: F[Option[Chunk[Byte]]],
       maxHeaderSize: Int,
       requestHeaderReceiveTimeout: Duration,
       httpApp: HttpApp[F],
-      errorHandler: Throwable => F[Response[F]]): F[(Request[F], Response[F], Stream[F, Byte])] =
+      errorHandler: Throwable => F[Response[F]],
+      requestVault: Vault): F[(Request[F], Response[F], Option[Array[Byte]])] =
     for {
-      tup <- Parser.Request.parser(maxHeaderSize, durationToFinite(requestHeaderReceiveTimeout))(
-        incoming)
-      (req, rest) = tup
+      (req, drain) <- Parser.Request.parser(
+        maxHeaderSize,
+        durationToFinite(requestHeaderReceiveTimeout))(head, read)
       resp <- httpApp
-        .run(req)
+        .run(req.withAttributes(requestVault))
         .handleErrorWith(errorHandler)
         .handleError(_ => serverFailure.covary[F])
+      rest <- drain // TODO: handle errors?
     } yield (req, resp, rest)
 
   private[internal] def send[F[_]: Concurrent](socket: Socket[F])(
@@ -175,40 +185,84 @@ private[server] object ServerHelpers {
       onWriteFailure: (Option[Request[F]], Response[F], Throwable) => F[Unit]
   ): Stream[F, Nothing] = {
     val _ = logger
-    Stream
-      .unfoldLoopEval(socket.reads(receiveBufferSize, durationToFinite(idleTimeout)))(s =>
-        runApp(s, maxHeaderSize, requestHeaderReceiveTimeout, httpApp, errorHandler).attempt.map {
-          case Right((req, resp, rest)) => (Right((req, resp)), Some(rest))
-          case Left(e) => (Left(e), None)
-        })
-      .evalMap {
-        case Right((req, resp)) =>
-          postProcessResponse(req, resp).map(resp => (req, resp).asRight[Throwable])
-        case other => other.pure[F]
-      }
-      .evalTap {
-        case Right((request, response)) =>
-          send(socket)(Some(request), response, idleTimeout, onWriteFailure)
-        case Left(err) =>
-          err match {
-            case req: Parser.Request.ReqPrelude.ParsePreludeError
-                if req == Parser.Request.ReqPrelude.emptyStreamError =>
-              Applicative[F].unit
-            case err =>
-              errorHandler(err)
-                .handleError(_ => serverFailure.covary[F])
-                .flatMap(send(socket)(None, _, idleTimeout, onWriteFailure))
-          }
-      }
-      .takeWhile {
-        case Left(_) => false
-        case Right((req, resp)) =>
-          !(
-            req.headers.get(Connection).exists(_.hasClose) ||
-              resp.headers.get(Connection).exists(_.hasClose)
-          )
-      }
-      .drain ++ Stream.eval(socket.close).drain
+    val read: F[Option[Chunk[Byte]]] = socket.read(receiveBufferSize, durationToFinite(idleTimeout))
+    Stream.eval(mkRequestVault(socket)).flatMap { requestVault =>
+      Stream
+        .unfoldLoopEval(Array.emptyByteArray)(incoming =>
+          runApp(
+            incoming,
+            read,
+            maxHeaderSize,
+            requestHeaderReceiveTimeout,
+            httpApp,
+            errorHandler,
+            requestVault).attempt.map {
+            case Right((req, resp, rest)) => (Right((req, resp)), rest)
+            case Left(e) => (Left(e), None)
+          })
+        .evalMap {
+          case Right((req, resp)) =>
+            postProcessResponse(req, resp).map(resp => (req, resp).asRight[Throwable])
+          case other => other.pure[F]
+        }
+        .evalTap {
+          case Right((request, response)) =>
+            send(socket)(Some(request), response, idleTimeout, onWriteFailure)
+          case Left(err) =>
+            err match {
+              case req: Parser.Request.ReqPrelude.ParsePreludeError
+                  if req == Parser.Request.ReqPrelude.emptyStreamError =>
+                Applicative[F].unit
+              case err =>
+                errorHandler(err)
+                  .handleError(_ => serverFailure.covary[F])
+                  .flatMap(send(socket)(None, _, idleTimeout, onWriteFailure))
+            }
+        }
+        .takeWhile {
+          case Left(_) => false
+          case Right((req, resp)) =>
+            !(
+              req.headers.get(Connection).exists(_.hasClose) ||
+                resp.headers.get(Connection).exists(_.hasClose)
+            )
+        }
+        .drain ++ Stream.eval(socket.close).drain
+    }
   }
 
+  private def mkRequestVault[F[_]: Applicative](socket: Socket[F]) =
+    (mkConnectionInfo(socket), mkSecureSession(socket)).mapN(_ ++ _)
+
+  private def mkConnectionInfo[F[_]: Apply](socket: Socket[F]) =
+    (socket.localAddress, socket.remoteAddress).mapN {
+      case (local: InetSocketAddress, remote: InetSocketAddress) =>
+        Vault.empty.insert(
+          Request.Keys.ConnectionInfo,
+          Request.Connection(
+            local = SocketAddress.fromInetSocketAddress(local),
+            remote = SocketAddress.fromInetSocketAddress(remote),
+            secure = socket.isInstanceOf[TLSSocket[F]]
+          )
+        )
+      case _ =>
+        Vault.empty
+    }
+
+  private def mkSecureSession[F[_]: Applicative](socket: Socket[F]) =
+    socket match {
+      case socket: TLSSocket[F] =>
+        socket.session
+          .map { session =>
+            (
+              Option(session.getId).map(ByteVector(_).toHex),
+              Option(session.getCipherSuite),
+              Option(session.getCipherSuite).map(deduceKeyLength),
+              Some(getCertChain(session))
+            ).mapN(SecureSession.apply)
+          }
+          .map(Vault.empty.insert(ServerRequestKeys.SecureSession, _))
+      case _ =>
+        Vault.empty.pure[F]
+    }
 }
diff --git a/website/src/hugo/content/changelog.md b/website/src/hugo/content/changelog.md
index 3e0b830217..c988470390 100644
--- a/website/src/hugo/content/changelog.md
+++ b/website/src/hugo/content/changelog.md
@@ -51,7 +51,11 @@ it.
 * play-json-2.10.0-RC1
 * scala-xml-2.0.0-M4
 
-# v0.21.19
+### Breaking changes
+
+* [#4371](https://github.com/http4s/http4s/pull/4371): Replace jawn-play with an internal copy of the facade to work around `withDottyCompat` issues.
+
+# v0.21.19 (2021-02-13)
 
 ## http4s-core
 
@@ -59,6 +63,24 @@ it.
 
 * [#4337](https://github.com/http4s/http4s/pull/4337): Deprecate `Header.Recurring.GetT`, which is unused
 
+## http4s-client
+
+### Bugfixes
+
+* [#4403](https://github.com/http4s/http4s/pull/4403): Remove `Content-Coding` and `Content-Length` headers after decompressing in the `GZip` middleware.
+
+## http4s-ember-core
+
+### Bugfixes
+
+* [#4348](https://github.com/http4s/http4s/pull/4348): Handle partially read bodies in persistent connections when the connection is recycled.
+
+## http4s-ember-server
+
+### Enhancements
+
+* [#4400](https://github.com/http4s/http4s/pull/4400): Implement the `ConnectionInfo` and `SecureSession` request vault attributes, for parity with the Blaze and Servlet backends
+
 ## http4s-argonaut
 
 * [#4366](https://github.com/http4s/http4s/pull/4370): Deprecate http4s-argonaut.  It won't be published starting in 0.22.


Test patch:
diff --git a/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSpec.scala b/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSpec.scala
deleted file mode 100644
index aed1d97ec9..0000000000
--- a/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSpec.scala
+++ /dev/null
@@ -1,221 +0,0 @@
-/*
- * Copyright 2019 http4s.org
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.http4s.ember.client.internal
-
-import cats.syntax.all._
-import cats.data.NonEmptyList
-import cats.effect._
-import org.http4s._
-import cats.effect.testing.specs2.CatsEffect
-import org.http4s.headers.{Connection, Date, `User-Agent`}
-import org.http4s.ember.client.EmberClientBuilder
-import org.typelevel.ci.CIString
-import org.typelevel.keypool.Reusable
-
-class ClientHelpersSpec extends Http4sSpec with CatsEffect {
-  "Request Preprocessing" should {
-    "add a date header if not present" in {
-      ClientHelpers
-        .preprocessRequest(Request[IO](), None)
-        .map { req =>
-          req.headers.get(Date) must beSome
-        }
-    }
-    "not add a date header if already present" in {
-      ClientHelpers
-        .preprocessRequest(
-          Request[IO](
-            headers = Headers.of(Date(HttpDate.Epoch))
-          ),
-          None)
-        .map { req =>
-          req.headers.get(Date) must beSome.like { case d: Date =>
-            d.date === HttpDate.Epoch
-          }
-        }
-    }
-    "add a connection keep-alive header if not present" in {
-      ClientHelpers
-        .preprocessRequest(Request[IO](), None)
-        .map { req =>
-          req.headers.get(Connection) must beSome.like { case c: Connection =>
-            c.hasKeepAlive must beTrue
-          }
-        }
-    }
-
-    "not add a connection header if already present" in {
-      ClientHelpers
-        .preprocessRequest(
-          Request[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
-          None
-        )
-        .map { req =>
-          req.headers.get(Connection) must beSome.like { case c: Connection =>
-            c.hasKeepAlive must beFalse
-          }
-        }
-    }
-
-    "add default user-agent" in {
-      ClientHelpers
-        .preprocessRequest(Request[IO](), EmberClientBuilder.default[IO].userAgent)
-        .map { req =>
-          req.headers.get(`User-Agent`) must beSome
-        }
-    }
-
-    "not change a present user-agent" in {
-      val name = "foo"
-      ClientHelpers
-        .preprocessRequest(
-          Request[IO](
-            headers = Headers.of(`User-Agent`(ProductId(name, None)))
-          ),
-          EmberClientBuilder.default[IO].userAgent)
-        .map { req =>
-          req.headers.get(`User-Agent`) must beSome.like { case e =>
-            e.product.value must_=== name
-          }
-        }
-    }
-  }
-
-  "Postprocess response" should {
-    "reuse when body is run" in {
-
-      for {
-        reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-        resp =
-          ClientHelpers
-            .postProcessResponse(
-              Request[IO](),
-              Response[IO](),
-              reuse
-            )
-        testResult <-
-          resp.body.compile.drain >>
-            reuse.get.map { case r =>
-              r must beEqualTo(Reusable.Reuse)
-            }
-      } yield testResult
-    }
-
-    "do not reuse when body is not run" in {
-      for {
-        reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-        _ =
-          ClientHelpers
-            .postProcessResponse(
-              Request[IO](),
-              Response[IO](),
-              reuse
-            )
-
-        testResult <-
-          reuse.get.map { case r =>
-            r must beEqualTo(Reusable.DontReuse)
-          }
-      } yield testResult
-    }
-
-    "do not reuse when error encountered running stream" in {
-      for {
-        reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-        resp =
-          ClientHelpers
-            .postProcessResponse(
-              Request[IO](),
-              Response[IO](body = fs2.Stream.raiseError[IO](new Throwable("Boo!"))),
-              reuse
-            )
-        testResult <-
-          resp.body.compile.drain.attempt >>
-            reuse.get.map { case r =>
-              r must beEqualTo(Reusable.DontReuse)
-            }
-      } yield testResult
-    }
-
-//    "do not reuse when cancellation encountered running stream" in {
-//      for {
-//        reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-//
-//        testResult <-
-//          ClientHelpers
-//            .postProcessResponse(
-//              Request[IO](),
-//              Response[IO](body = fs2
-//                .Stream(1, 2, 3, 4, 5)
-//                .map(_.toByte)
-//                .zipLeft(
-//                  fs2.Stream.awakeDelay[IO](1.second)
-//                )
-//                .interruptAfter(2.seconds)),
-//              reuse
-//            )
-//            .use { resp =>
-//              resp.body.compile.drain.attempt >>
-//                reuse.get.map { case r =>
-//                  r must beEqualTo(Reusable.DontReuse)
-//                }
-//            }
-//      } yield testResult
-//    }.pendingUntilFixed
-
-    "do not reuse when connection close is set on request" in {
-      for {
-        reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-        resp =
-          ClientHelpers
-            .postProcessResponse(
-              Request[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
-              Response[IO](),
-              reuse
-            )
-        testResult <-
-          resp.body.compile.drain >>
-            reuse.get.map { case r =>
-              r must beEqualTo(Reusable.DontReuse)
-            }
-      } yield testResult
-    }
-
-    "do not reuse when connection close is set on response" in {
-      for {
-        reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-        resp =
-          ClientHelpers
-            .postProcessResponse(
-              Request[IO](),
-              Response[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
-              reuse
-            )
-        testResult <-
-          resp.body.compile.drain >>
-            reuse.get.map { case r =>
-              r must beEqualTo(Reusable.DontReuse)
-            }
-      } yield testResult
-    }
-  }
-}
diff --git a/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSuite.scala b/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSuite.scala
index 0389d1e80c..4fe40f7dc3 100644
--- a/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSuite.scala
+++ b/ember-client/src/test/scala/org/http4s/ember/client/internal/ClientHelpersSuite.scala
@@ -16,7 +16,6 @@
 
 package org.http4s.ember.client.internal
 
-import cats.syntax.all._
 import cats.data.NonEmptyList
 import cats.effect._
 import org.http4s._
@@ -24,7 +23,6 @@ import org.http4s.headers.{Connection, Date, `User-Agent`}
 import org.http4s.ember.client.EmberClientBuilder
 import org.typelevel.ci.CIString
 import org.typelevel.keypool.Reusable
-import scala.concurrent.duration._
 
 class ClientHelpersSuite extends Http4sSuite {
 
@@ -101,127 +99,92 @@ class ClientHelpersSuite extends Http4sSuite {
       .assertEquals(Some(name))
   }
 
-  test("Postprocess response should reuse when body is run") {
-
+  test("Postprocess response should reuse") {
     for {
+      nextBytes <- Ref[IO].of(Array.emptyByteArray)
       reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
 
-      resp =
-        ClientHelpers
-          .postProcessResponse(
-            Request[IO](),
-            Response[IO](),
-            reuse
-          )
-      testResult <-
-        resp.body.compile.drain >>
-          reuse.get.map { case r =>
-            assertEquals(r, Reusable.Reuse)
-          }
-    } yield testResult
-  }
-
-  test("Postprocess response should do not reuse when body is not run") {
-    for {
-      reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-      _ =
-        ClientHelpers
-          .postProcessResponse(
-            Request[IO](),
-            Response[IO](),
-            reuse
-          )
-
-      testResult <-
-        reuse.get.map { case r =>
-          assertEquals(r, Reusable.DontReuse)
-        }
+      _ <- ClientHelpers
+        .postProcessResponse[IO](
+          Request[IO](),
+          Response[IO](),
+          IO.pure(Some(Array.emptyByteArray)),
+          nextBytes,
+          reuse
+        )
+      testResult <- reuse.get.map { case r =>
+        assertEquals(r, Reusable.Reuse)
+      }
     } yield testResult
   }
 
-  test("Postprocess response should do not reuse when error encountered running stream") {
+  test("Postprocess response should save drained bytes when reused") {
     for {
+      nextBytes <- Ref[IO].of(Array.emptyByteArray)
       reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
 
-      resp =
-        ClientHelpers
-          .postProcessResponse(
-            Request[IO](),
-            Response[IO](body = fs2.Stream.raiseError[IO](new Throwable("Boo!"))),
-            reuse
-          )
-      testResult <-
-        resp.body.compile.drain.attempt >>
-          reuse.get.map { case r =>
-            assertEquals(r, Reusable.DontReuse)
-          }
-    } yield testResult
+      _ <- ClientHelpers.postProcessResponse[IO](
+        Request[IO](),
+        Response[IO](),
+        IO.pure(Some(Array[Byte](1, 2, 3))),
+        nextBytes,
+        reuse
+      )
+      drained <- nextBytes.get
+    } yield assertEquals(drained.toList, List[Byte](1, 2, 3))
   }
 
-  // pending
-  test(
-    "Postprocess response should do not reuse when cancellation encountered running stream".fail) {
+  test("Postprocess response should not reuse when connection close is set on request") {
     for {
+      nextBytes <- Ref[IO].of(Array.emptyByteArray)
       reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-      resp =
-        ClientHelpers
-          .postProcessResponse(
-            Request[IO](),
-            Response[IO](body = fs2
-              .Stream(1, 2, 3, 4, 5)
-              .map(_.toByte)
-              .zipLeft(
-                fs2.Stream.awakeDelay[IO](1.second)
-              )
-              .interruptAfter(2.seconds)),
-            reuse
-          )
-      testResult <-
-        resp.body.compile.drain.attempt >>
-          reuse.get.map { case r =>
-            assertEquals(r, Reusable.DontReuse)
-          }
+      _ <- ClientHelpers
+        .postProcessResponse[IO](
+          Request[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
+          Response[IO](),
+          IO.pure(Some(Array.emptyByteArray)),
+          nextBytes,
+          reuse
+        )
+      testResult <- reuse.get.map { case r =>
+        assertEquals(r, Reusable.DontReuse)
+      }
     } yield testResult
   }
 
-  test("Postprocess response should do not reuse when connection close is set on request") {
+  test("Postprocess response should do not reuse when connection close is set on response") {
     for {
+      nextBytes <- Ref[IO].of(Array.emptyByteArray)
       reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-      resp =
-        ClientHelpers
-          .postProcessResponse[IO](
-            Request[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
-            Response[IO](),
-            reuse
-          )
-      testResult <-
-        resp.body.compile.drain >>
-          reuse.get.map { case r =>
-            assertEquals(r, Reusable.DontReuse)
-          }
+      _ <- ClientHelpers
+        .postProcessResponse[IO](
+          Request[IO](),
+          Response[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
+          IO.pure(Some(Array.emptyByteArray)),
+          nextBytes,
+          reuse
+        )
+      testResult <- reuse.get.map { case r =>
+        assertEquals(r, Reusable.DontReuse)
+      }
     } yield testResult
   }
 
-  test("Postprocess response should do not reuse when connection close is set on response") {
+  test("Postprocess response should do not reuse when drain is None") {
     for {
+      nextBytes <- Ref[IO].of(Array.emptyByteArray)
       reuse <- Ref[IO].of(Reusable.DontReuse: Reusable)
-
-      resp =
-        ClientHelpers
-          .postProcessResponse(
-            Request[IO](),
-            Response[IO](headers = Headers.of(Connection(NonEmptyList.of(CIString("close"))))),
-            reuse
-          )
-      testResult <-
-        resp.body.compile.drain >>
-          reuse.get.map { case r =>
-            assertEquals(r, Reusable.DontReuse)
-          }
+      _ <- ClientHelpers
+        .postProcessResponse[IO](
+          Request[IO](),
+          Response[IO](),
+          IO.pure(None),
+          nextBytes,
+          reuse
+        )
+      testResult <- reuse.get.map { case r =>
+        assertEquals(r, Reusable.DontReuse)
+      }
     } yield testResult
   }
-
 }
diff --git a/ember-core/src/test/scala/org/http4s/ember/core/ParsingSpec.scala b/ember-core/src/test/scala/org/http4s/ember/core/ParsingSpec.scala
index c714f7ef91..9d5e62ba19 100644
--- a/ember-core/src/test/scala/org/http4s/ember/core/ParsingSpec.scala
+++ b/ember-core/src/test/scala/org/http4s/ember/core/ParsingSpec.scala
@@ -16,6 +16,7 @@
 
 package org.http4s.ember.core
 
+import cats.effect.std.Queue
 import org.http4s._
 import org.http4s.implicits._
 import scodec.bits.ByteVector
@@ -31,6 +32,13 @@ class ParsingSpec extends Http4sSuite {
     def stripLines(s: String): String = s.replace("\r\n", "\n")
     def httpifyString(s: String): String = s.replace("\n", "\r\n")
 
+    def taking[F[_]: Concurrent, A](stream: Stream[F, A]): F[F[Option[Chunk[A]]]] =
+      for {
+        q <- Queue.unbounded[F, Option[Chunk[A]]]
+        _ <- stream.chunks.map(Some(_)).evalMap(q.offer(_)).compile.drain.void
+        _ <- q.offer(None)
+      } yield q.take
+
     // Only for Use with Text Requests
     def parseRequestRig[F[_]: Concurrent: Temporal](s: String): F[Request[F]] = {
       val byteStream: Stream[F, Byte] = Stream
@@ -39,7 +47,9 @@ class ParsingSpec extends Http4sSuite {
         .map(httpifyString)
         .through(fs2.text.utf8Encode[F])
 
-      Parser.Request.parser[F](Int.MaxValue, None)(byteStream).map(_._1)
+      taking(byteStream).flatMap { read =>
+        Parser.Request.parser[F](Int.MaxValue, None)(Array.emptyByteArray, read).map(_._1)
+      }
     }
 
     def parseResponseRig[F[_]: Concurrent: Temporal](s: String): Resource[F, Response[F]] = {
@@ -49,8 +59,11 @@ class ParsingSpec extends Http4sSuite {
         .map(httpifyString)
         .through(fs2.text.utf8Encode[F])
 
-      val action = Parser.Response.parser[F](Int.MaxValue, None)(byteStream).map(_._1) //(logger)
-      Resource.eval(action)
+      Resource.eval(
+        taking(byteStream).flatMap { read =>
+          Parser.Response.parser[F](Int.MaxValue, None)(Array.emptyByteArray, read).map(_._1)
+        }
+      )
     }
 
     def forceScopedParsing[F[_]: Concurrent](s: String): Stream[F, Byte] = {
@@ -165,16 +178,19 @@ class ParsingSpec extends Http4sSuite {
 
     val raw2 = """}
           |""".stripMargin
+
     val http1 = Helpers.httpifyString(raw1)
 
     val http2 = Helpers.httpifyString(raw2)
     val encoded = (Stream(http1) ++ Stream(http2)).through(fs2.text.utf8Encode)
 
     (for {
+      take <- Helpers.taking[IO, Byte](encoded)
       parsed <-
         Parser.Response
           .parser[IO](defaultMaxHeaderLength, None)(
-            encoded
+            Array.emptyByteArray,
+            take
             //Helpers.forceScopedParsing[IO](raw) // Cuts off `}` in current test. Why?
             // I don't follow what the rig is testing vs this.
           ) //(logger)
@@ -184,21 +200,67 @@ class ParsingSpec extends Http4sSuite {
     } yield parsed == "{}").assert
   }
 
+  test(
+    "Parser.Request.parser should parse a request with Content-Length and return the rest of the stream") {
+    val raw =
+      """POST /foo HTTP/1.1
+          |Host: localhost:8080
+          |User-Agent: curl/7.64.1
+          |Accept: */*
+          |Content-Length: 5
+          |
+          |helloeverything after the body""".stripMargin
+
+    val byteStream = Stream
+      .emit(raw)
+      .covary[IO]
+      .map(Helpers.httpifyString)
+      .through(text.utf8Encode)
+
+    (for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      result <- Parser.Request.parser[IO](Int.MaxValue, None)(Array.emptyByteArray, take)
+      body <- result._1.body.through(text.utf8Decode).compile.string
+      rest <- Stream
+        .eval(result._2)
+        .flatMap(chunk => Stream.chunk(Chunk.byteVector(ByteVector(chunk.get))))
+        .through(text.utf8Decode)
+        .compile
+        .string
+    } yield body == "hello" && rest == "everything after the body").assert
+  }
+
+  test("Parser.Request.parser should parse two requests in a row") {
+    val reqS =
+      Stream(
+        "GET /foo HTTP/1.1\r\n",
+        "Accept: text/plain\r\n\r\nGET /foo HTTP/1.1\r\n",
+        "Accept: text/plain\r\n\r\n"
+      )
+    val byteStream: Stream[IO, Byte] = reqS
+      .flatMap(s =>
+        Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.ISO_8859_1))))
+
+    (for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      req1 <- Parser.Request.parser[IO](Int.MaxValue, None)(Array.emptyByteArray, take)
+      drained <- req1._2
+      req2 <- Parser.Request.parser[IO](Int.MaxValue, None)(drained.get, take)
+    } yield req1._1.method == Method.GET && req2._1.method == Method.GET).assert
+  }
+
   test("Parser.Response.parser should handle a chunked response") {
     val defaultMaxHeaderLength = 4096
     val base =
       "SFRUUC8xLjEgMjAwIE9LDQpBcGktVmVyc2lvbjogMS40MA0KQ29udGVudC1UeXBlOiBhcHBsaWNhdGlvbi9qc29uDQpEb2NrZXItRXhwZXJpbWVudGFsOiBmYWxzZQ0KT3N0eXBlOiBsaW51eA0KU2VydmVyOiBEb2NrZXIvMTkuMDMuMTEtY2UgKGxpbnV4KQ0KRGF0ZTogRnJpLCAyNiBKdW4gMjAyMCAyMjozNTo0MiBHTVQNClRyYW5zZmVyLUVuY29kaW5nOiBjaHVua2VkDQoNCjhjMw0KeyJJRCI6IllNS0U6MkZZMzpTUUc3OjZSSFo6TFlTVDpRUk9JOkU1NEU6UTdXRjpERElLOlNOSUE6Rk5UTzpJVllSIiwiQ29udGFpbmVycyI6MjUsIkNvbnRhaW5lcnNSdW5uaW5nIjowLCJDb250YWluZXJzUGF1c2VkIjowLCJDb250YWluZXJzU3RvcHBlZCI6MjUsIkltYWdlcyI6ODMsIkRyaXZlciI6Im92ZXJsYXkyIiwiRHJpdmVyU3RhdHVzIjpbWyJCYWNraW5nIEZpbGVzeXN0ZW0iLCJleHRmcyJdLFsiU3VwcG9ydHMgZF90eXBlIiwidHJ1ZSJdLFsiTmF0aXZlIE92ZXJsYXkgRGlmZiIsImZhbHNlIl1dLCJTeXN0ZW1TdGF0dXMiOm51bGwsIlBsdWdpbnMiOnsiVm9sdW1lIjpbImxvY2FsIl0sIk5ldHdvcmsiOlsiYnJpZGdlIiwiaG9zdCIsImlwdmxhbiIsIm1hY3ZsYW4iLCJudWxsIiwib3ZlcmxheSJdLCJBdXRob3JpemF0aW9uIjpudWxsLCJMb2ciOlsiYXdzbG9ncyIsImZsdWVudGQiLCJnY3Bsb2dzIiwiZ2VsZiIsImpvdXJuYWxkIiwianNvbi1maWxlIiwibG9jYWwiLCJsb2dlbnRyaWVzIiwic3BsdW5rIiwic3lzbG9nIl19LCJNZW1vcnlMaW1pdCI6dHJ1ZSwiU3dhcExpbWl0Ijp0cnVlLCJLZXJuZWxNZW1vcnkiOnRydWUsIktlcm5lbE1lbW9yeVRDUCI6dHJ1ZSwiQ3B1Q2ZzUGVyaW9kIjp0cnVlLCJDcHVDZnNRdW90YSI6dHJ1ZSwiQ1BVU2hhcmVzIjp0cnVlLCJDUFVTZXQiOnRydWUsIlBpZHNMaW1pdCI6dHJ1ZSwiSVB2NEZvcndhcmRpbmciOnRydWUsIkJyaWRnZU5mSXB0YWJsZXMiOnRydWUsIkJyaWRnZU5mSXA2dGFibGVzIjp0cnVlLCJEZWJ1ZyI6ZmFsc2UsIk5GZCI6MjQsIk9vbUtpbGxEaXNhYmxlIjp0cnVlLCJOR29yb3V0aW5lcyI6NDAsIlN5c3RlbVRpbWUiOiIyMDIwLTA2LTI2VDE1OjM1OjQyLjU1MjUzMzQzMS0wNzowMCIsIkxvZ2dpbmdEcml2ZXIiOiJqc29uLWZpbGUiLCJDZ3JvdXBEcml2ZXIiOiJjZ3JvdXBmcyIsIk5FdmVudHNMaXN0ZW5lciI6MCwiS2VybmVsVmVyc2lvbiI6IjUuNy42LWFyY2gxLTEiLCJPcGVyYXRpbmdTeXN0ZW0iOiJBcmNoIExpbnV4IiwiT1NUeXBlIjoibGludXgiLCJBcmNoaXRlY3R1cmUiOiJ4ODZfNjQiLCJJbmRleFNlcnZlckFkZHJlc3MiOiJodHRwczovL2luZGV4LmRvY2tlci5pby92MS8iLCJSZWdpc3RyeUNvbmZpZyI6eyJBbGxvd05vbmRpc3RyaWJ1dGFibGVBcnRpZmFjdHNDSURScyI6W10sIkFsbG93Tm9uZGlzdHJpYnV0YWJsZUFydGlmYWN0c0hvc3RuYW1lcyI6W10sIkluc2VjdXJlUmVnaXN0cnlDSURScyI6WyIxMjcuMC4wLjAvOCJdLCJJbmRleENvbmZpZ3MiOnsiZG9ja2VyLmlvIjp7Ik5hbWUiOiJkb2NrZXIuaW8iLCJNaXJyb3JzIjpbXSwiU2VjdXJlIjp0cnVlLCJPZmZpY2lhbCI6dHJ1ZX19LCJNaXJyb3JzIjpbXX0sIk5DUFUiOjQsIk1lbVRvdGFsIjo4MjIwOTgzMjk2LCJHZW5lcmljUmVzb3VyY2VzIjpudWxsLCJEb2NrZXJSb290RGlyIjoiL3Zhci9saWIvZG9ja2VyIiwiSHR0cFByb3h5IjoiIiwiSHR0cHNQcm94eSI6IiIsIk5vUHJveHkiOiIiLCJOYW1lIjoiZGF2ZW5wb3J0LWxhcHRvcCIsIkxhYmVscyI6W10sIkV4cGVyaW1lbnRhbEJ1aWxkIjpmYWxzZSwiU2VydmVyVmVyc2lvbiI6IjE5LjAzLjExLWNlIiwiQ2x1c3RlclN0b3JlIjoiIiwiQ2x1c3RlckFkdmVydGlzZSI6IiIsIlJ1bnRpbWVzIjp7InJ1bmMiOnsicGF0aCI6InJ1bmMifX0sIkRlZmF1bHRSdW50aW1lIjoicnVuYyIsIlN3YXJtIjp7Ik5vZGVJRCI6IiIsIk5vZGVBZGRyIjoiIiwiTG9jYWxOb2RlU3RhdGUiOiJpbmFjdGl2ZSIsIkNvbnRyb2xBdmFpbGFibGUiOmZhbHNlLCJFcnJvciI6IiIsIlJlbW90ZU1hbmFnZXJzIjpudWxsfSwiTGl2ZVJlc3RvcmVFbmFibGVkIjpmYWxzZSwiSXNvbGF0aW9uIjoiIiwiSW5pdEJpbmFyeSI6ImRvY2tlci1pbml0IiwiQ29udGFpbmVyZENvbW1pdCI6eyJJRCI6ImQ3NmMxMjFmNzZhNWZjOGE0NjJkYzY0NTk0YWVhNzJmZTE4ZTExNzgubSIsIkV4cGVjdGVkIjoiZDc2YzEyMWY3NmE1ZmM4YTQ2MmRjNjQ1OTRhZWE3MmZlMThlMTE3OC5tIn0sIlJ1bmNDb21taXQiOnsiSUQiOiJkYzkyMDhhMzMwM2ZlZWY1YjM4MzlmNDMyM2Q5YmViMzZkZjBhOWRkIiwiRXhwZWN0ZWQiOiJkYzkyMDhhMzMwM2ZlZWY1YjM4MzlmNDMyM2Q5YmViMzZkZjBhOWRkIn0sIkluaXRDb21taXQiOnsiSUQiOiJmZWMzNjgzIiwiRXhwZWN0ZWQiOiJmZWMzNjgzIn0sIlNlY3VyaXR5T3B0aW9ucyI6WyJuYW1lPXNlY2NvbXAscHJvZmlsZT1kZWZhdWx0Il0sIldhcm5pbmdzIjpudWxsfQoNCjANCg0K"
     val baseBv = ByteVector.fromBase64(base).get
 
-    Parser.Response
-      .parser[IO](defaultMaxHeaderLength, None)(Stream.chunk(Chunk.byteVector(baseBv)))
-      .flatMap { case (resp, _) =>
-        resp.body.through(text.utf8Decode).compile.string
-      }
-      .map { x =>
-        x.size > 0
-      }
-      .assert
+    (for {
+      take <- Helpers.taking[IO, Byte](Stream.chunk(Chunk.byteVector(baseBv)))
+      result <- Parser.Response
+        .parser[IO](defaultMaxHeaderLength, None)(Array.emptyByteArray, take)
+      body <- result._1.body.through(text.utf8Decode).compile.string
+    } yield body.size > 0).assert
   }
 
   test("Parser.Response.parser should parse a chunked simple") {
@@ -223,14 +285,12 @@ class ParsingSpec extends Http4sSuite {
       .flatMap(s =>
         Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.ISO_8859_1))))
 
-    Parser.Response
-      .parser[IO](defaultMaxHeaderLength, None)(byteStream)
-      .flatMap { case (resp, _) =>
-        resp.body.through(text.utf8Decode).compile.string.map { body =>
-          body == "MozillaDeveloperNetwork"
-        }
-      }
-      .assert
+    (for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      resp <- Parser.Response
+        .parser[IO](defaultMaxHeaderLength, None)(Array.emptyByteArray, take)
+      body <- resp._1.body.through(text.utf8Decode).compile.string
+    } yield body == "MozillaDeveloperNetwork").assert
   }
 
   test("Parser.Response.parser should parse a chunked with trailer headers") {
@@ -251,21 +311,47 @@ class ParsingSpec extends Http4sSuite {
         "Expires: Wed, 21 Oct 2015 07:28:00 GMT\r\n\r\n"
         // "\r\n"
       )
+
     val byteStream: Stream[IO, Byte] = respS
       .flatMap(s =>
         Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.US_ASCII))))
 
-    Parser.Response
-      .parser[IO](defaultMaxHeaderLength, None)(byteStream)
-      .flatMap { case (resp, _) =>
-        for {
-          body <- resp.body.through(text.utf8Decode).compile.string
-          trailers <- resp.trailerHeaders
-        } yield (body == "MozillaDeveloperNetwork").&&(
-          trailers.get(Expires).isDefined
-        )
-      }
-      .assert
+    (for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      result <- Parser.Response
+        .parser[IO](defaultMaxHeaderLength, None)(Array.emptyByteArray, take)
+      body <- result._1.body.through(text.utf8Decode).compile.string
+      trailers <- result._1.trailerHeaders
+    } yield body == "MozillaDeveloperNetwork" && trailers.get(Expires).isDefined).assert
+  }
+
+  test(
+    "Parser.Response.parser should parse a response with Content-Length and return the rest of the stream") {
+    val defaultMaxHeaderLength = 4096
+    val raw =
+      """HTTP/1.1 200 OK
+        |Content-Length: 5
+        |
+        |helloeverything after the body""".stripMargin
+
+    val byteStream = Stream
+      .emit(raw)
+      .covary[IO]
+      .map(Helpers.httpifyString)
+      .through(text.utf8Encode)
+
+    (for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      result <- Parser.Response
+        .parser[IO](defaultMaxHeaderLength, None)(Array.emptyByteArray, take)
+      body <- result._1.body.through(text.utf8Decode).compile.string
+      rest <- Stream
+        .eval(result._2)
+        .flatMap(chunk => Stream.chunk(Chunk.byteVector(ByteVector(chunk.get))))
+        .through(text.utf8Decode)
+        .compile
+        .string
+    } yield body == "hello" && rest == "everything after the body").assert
   }
 
   test("Header Parser should handle headers in a section") {
@@ -296,7 +382,6 @@ class ParsingSpec extends Http4sSuite {
     assert(
       length == Some(11L)
     )
-
   }
 
   test("Header Parser should Handle weird chunking") {
@@ -312,17 +397,14 @@ class ParsingSpec extends Http4sSuite {
       .flatMap(s =>
         Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.US_ASCII))))
 
-    val headers = Parser.HeaderP
-      .parseHeaders(byteStream, defaultMaxHeaderLength, None)
-      .flatMap { case (headers, _, _, _) =>
-        Pull.output1(headers)
-      }
-      .stream
-      .compile
-      .lastOrError
-      .map(_.toList)
+    val result = for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      headers <- Parser.HeaderP
+        .parseHeaders(Array.emptyByteArray, take, defaultMaxHeaderLength, None)
+        .map(_._1)
+    } yield headers.toList
 
-    headers.assertEquals(
+    result.assertEquals(
       List(
         Header("Content-Type", "text/plain"),
         Header("Transfer-Encoding", "chunked"),
@@ -368,4 +450,99 @@ class ParsingSpec extends Http4sSuite {
     }
   }
 
+  test("Parser.Response.parser should parse two responses in a row") {
+    val defaultMaxHeaderLength = 4096
+    val respS =
+      Stream(
+        "HTTP/1.1 200 OK\r\n",
+        "Content-Type: text/plain\r\n",
+        "Content-Length: 5\r\n\r\n",
+        "helloHTTP/1.1 200 OK\r\n", // this is the crucial part
+        "Content-Type: text/plain\r\n",
+        "Content-Length: 5\r\n\r\nworld"
+      )
+    val byteStream: Stream[IO, Byte] = respS
+      .flatMap(s =>
+        Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.ISO_8859_1))))
+
+    (for {
+      take <- Helpers.taking[IO, Byte](byteStream)
+      resp1 <- Parser.Response
+        .parser[IO](defaultMaxHeaderLength, None)(Array.emptyByteArray, take)
+      body1 <- resp1._1.body.through(fs2.text.utf8Decode).compile.string
+      drained <- resp1._2
+      resp2 <- Parser.Response
+        .parser[IO](defaultMaxHeaderLength, None)(drained.get, take)
+      body2 <- resp2._1.body.through(fs2.text.utf8Decode).compile.string
+    } yield body1 == "hello" && body2 == "world").assert
+  }
+
+  test("Parser.Body should completely consume a body") {
+    val bodyS = Stream("hello ", "world!")
+    val byteStream: Stream[IO, Byte] = bodyS
+      .flatMap(s =>
+        Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.US_ASCII))))
+
+    for {
+      take <- Helpers.taking(byteStream)
+      body <- Parser.Body.parseFixedBody(12L, Array.emptyByteArray, take)
+      bodyString <- body._1.through(fs2.text.utf8Decode).compile.string
+      drained <- body._2
+    } yield {
+      assertEquals(bodyString, "hello world!")
+      assertEquals(drained.map(_.toList), Some(Nil))
+    }
+  }
+
+  test("Parser.Body should partially consume a body") {
+    val bodyS = Stream("hello ", "world!")
+    val byteStream: Stream[IO, Byte] = bodyS
+      .flatMap(s =>
+        Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.US_ASCII))))
+
+    for {
+      take <- Helpers.taking(byteStream)
+      body <- Parser.Body.parseFixedBody(12L, Array.emptyByteArray, take)
+      bodyString <- body._1.take(2).through(fs2.text.utf8Decode).compile.string
+      drained <- body._2
+    } yield {
+      assertEquals(bodyString, "he")
+      assertEquals(drained, None)
+    }
+  }
+
+  test("Parser.Body should consume previous bytes") {
+    val bodyS = Stream("world!")
+    val byteStream: Stream[IO, Byte] = bodyS
+      .flatMap(s =>
+        Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.US_ASCII))))
+
+    for {
+      take <- Helpers.taking(byteStream)
+      body <- Parser.Body.parseFixedBody(
+        12L,
+        "hello ".getBytes(java.nio.charset.StandardCharsets.US_ASCII),
+        take)
+      bodyString <- body._1.through(fs2.text.utf8Decode).compile.string
+      drained <- body._2
+    } yield {
+      assertEquals(bodyString, "hello world!")
+      assertEquals(drained.map(_.toList), Some(Nil))
+    }
+  }
+
+  // Compiling a stream is generally undefined behavior but that doesn't stop us from testing it
+  test("Parser.Body should raise an error when compiling the body more than once") {
+    val bodyS = Stream("hello world!")
+    val byteStream: Stream[IO, Byte] = bodyS
+      .flatMap(s =>
+        Stream.chunk(Chunk.array(s.getBytes(java.nio.charset.StandardCharsets.US_ASCII))))
+
+    for {
+      take <- Helpers.taking(byteStream)
+      body <- Parser.Body.parseFixedBody(12L, Array.emptyByteArray, take)
+      _ <- body._1.compile.drain
+      _ <- body._1.compile.drain.intercept[Throwable]
+    } yield ()
+  }
 }