Repository: http4s/http4s
Pull request #6618 (2022-08-17 11:28:08)

Ember-Core / H2Client /  fromSocket: split for comprehension.
Redo of #6523, in a series of small-step commits. 


__________ 

Small refactor of the code in the `fromSocket` method of the `H2Client` class, for creating and starting up an `H2Connection`. Splits long code of the `for` comprehension (in Resource) into a set of auxiliary methods in `F`. 

Each of the auxiliary methods operates at the level of the "F" effect, and they are only lift to a `Resource` for the purposes of background running.

Initialising the `H2Connection` object is done in a separate method. Instead of accessing the components from the variables in the for loop, the other auxiliary methods access them
through the `H2Connection` fields themselves.




Patch:
diff --git a/build.sbt b/build.sbt
index e5bcc2b436..2cf7e48179 100644
--- a/build.sbt
+++ b/build.sbt
@@ -439,9 +439,6 @@ lazy val emberCore = libraryCrossProject("ember-core", CrossType.Full)
       ProblemFilters.exclude[DirectMissingMethodProblem](
         "org.http4s.ember.core.Parser#Response#RespPrelude.parsePrelude"
       ),
-      ProblemFilters.exclude[DirectMissingMethodProblem](
-        "org.http4s.ember.core.Parser#MessageP.recurseFind"
-      ),
       ProblemFilters.exclude[MissingClassProblem]("org.http4s.ember.core.EmptyStreamError"),
       ProblemFilters.exclude[MissingClassProblem]("org.http4s.ember.core.EmptyStreamError$"),
       ProblemFilters.exclude[MissingClassProblem]("org.http4s.ember.core.Parser$MessageP"),
diff --git a/core/shared/src/main/scala/org/http4s/Uri.scala b/core/shared/src/main/scala/org/http4s/Uri.scala
index 5fbce2a39d..1ceb9a0f87 100644
--- a/core/shared/src/main/scala/org/http4s/Uri.scala
+++ b/core/shared/src/main/scala/org/http4s/Uri.scala
@@ -54,7 +54,7 @@ import java.nio.CharBuffer
 import java.nio.charset.StandardCharsets
 import java.nio.charset.{Charset => JCharset}
 import scala.collection.immutable
-import scala.util.hashing.MurmurHash3
+import scala.math.Ordered
 
 /** Representation of the [[Request]] URI
   *
@@ -340,11 +340,10 @@ object Uri extends UriPlatform {
       this.segments == path.segments && path.absolute == this.absolute && path.endsWithSlash == this.endsWithSlash
 
     override def hashCode(): Int = {
-      var hash = Path.hashSeed
-      hash = MurmurHash3.mix(hash, segments.##)
-      hash = MurmurHash3.mix(hash, absolute.##)
-      hash = MurmurHash3.mix(hash, endsWithSlash.##)
-      MurmurHash3.finalizeHash(hash, 3)
+      var hash = segments.hashCode()
+      hash += 31 * absolute.hashCode()
+      hash += 31 * endsWithSlash.hashCode()
+      hash
     }
 
     def render(writer: Writer): writer.type = {
@@ -458,9 +457,6 @@ object Uri extends UriPlatform {
     lazy val Asterisk: Path =
       new Path(Vector(Segment("*")), absolute = false, endsWithSlash = false)
 
-    private val hashSeed: Int =
-      MurmurHash3.mix(MurmurHash3.productSeed, "Uri.Path".##)
-
     final class Segment private (val encoded: String) {
       def isEmpty = encoded.isEmpty
 
@@ -577,9 +573,8 @@ object Uri extends UriPlatform {
           )
       }
 
-    def http4sInstancesForPath: Order[Path] with Semigroup[Path] = http4sInstancesForPathBinCompat
-    implicit val http4sInstancesForPathBinCompat: Order[Path] with Semigroup[Path] with Hash[Path] =
-      new Order[Path] with Semigroup[Path] with Hash[Path] {
+    implicit val http4sInstancesForPath: Order[Path] with Semigroup[Path] =
+      new Order[Path] with Semigroup[Path] {
         def compare(x: Path, y: Path): Int = {
           def comparePaths[A: Order](focus: Path => A): Int =
             compareField(x, y, focus)
@@ -591,8 +586,6 @@ object Uri extends UriPlatform {
         }
 
         def combine(x: Path, y: Path): Path = x.concat(y)
-
-        def hash(x: Path): Int = x.##
       }
 
   }
diff --git a/ember-core/shared/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala b/ember-core/shared/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala
index ee3b2bf765..a88389d3aa 100644
--- a/ember-core/shared/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala
+++ b/ember-core/shared/src/main/scala/org/http4s/ember/core/ChunkedEncoding.scala
@@ -131,11 +131,9 @@ private[ember] object ChunkedEncoding {
       }
     } else {
       Parser.MessageP
-        .recurseFind(buffer, read, maxHeaderSize, Parser.HeaderP.ParserState.initial)(
-          (state, buffer) => Parser.HeaderP.parse(buffer, maxHeaderSize, state)
-        )(
-          _.idx
-        )
+        .recurseFind(buffer, read, maxHeaderSize)(buffer =>
+          Parser.HeaderP.parseHeaders(buffer, 0, maxHeaderSize)
+        )(_.idx)
         .map { case (headerP, rest) => Trailers(headerP.headers, rest) }
     }
 
diff --git a/ember-core/shared/src/main/scala/org/http4s/ember/core/Parser.scala b/ember-core/shared/src/main/scala/org/http4s/ember/core/Parser.scala
index 0b6f8f7fdd..6a8d97466a 100644
--- a/ember-core/shared/src/main/scala/org/http4s/ember/core/Parser.scala
+++ b/ember-core/shared/src/main/scala/org/http4s/ember/core/Parser.scala
@@ -16,6 +16,7 @@
 
 package org.http4s.ember.core
 import cats._
+import cats.data.EitherT
 import cats.effect.kernel.Concurrent
 import cats.effect.kernel.Deferred
 import cats.effect.kernel.Ref
@@ -31,23 +32,18 @@ private[ember] object Parser {
 
   object MessageP {
 
-    def recurseFind[F[_], S, A](
-        currentBuffer: Array[Byte],
-        read: Read[F],
-        maxHeaderSize: Int,
-        state: S,
-    )(
-        f: (S, Array[Byte]) => F[Either[S, A]]
+    def recurseFind[F[_], A](currentBuffer: Array[Byte], read: Read[F], maxHeaderSize: Int)(
+        f: Array[Byte] => F[Either[Unit, A]]
     )(idx: A => Int)(implicit F: MonadThrow[F]): F[(A, Array[Byte])] =
-      f(state, currentBuffer).flatMap {
-        case Left(s) =>
+      f(currentBuffer).flatMap {
+        case Left(_) =>
           if (currentBuffer.length > maxHeaderSize)
             F.raiseError(EmberException.MessageTooLong(maxHeaderSize))
           else {
             read.flatMap {
               case Some(chunk) =>
                 val nextBuffer = appendByteArrays(currentBuffer, chunk)
-                recurseFind(nextBuffer, read, maxHeaderSize, s)(f)(idx)
+                recurseFind(nextBuffer, read, maxHeaderSize)(f)(idx)
               case None if currentBuffer.length > 0 =>
                 F.raiseError(EmberException.ReachedEndOfStream())
               case _ => F.raiseError(EmberException.EmptyStream())
@@ -67,50 +63,26 @@ private[ember] object Parser {
   )
 
   object HeaderP {
+
     private[this] val colon: Byte = ':'.toByte
     private[this] val contentLengthS = "Content-Length"
     private[this] val transferEncodingS = "Transfer-Encoding"
     private[this] val chunkedS = "chunked"
 
-    final case class ParserState(
-        idx: Int,
-        state: Boolean,
-        throwable: Option[Throwable],
-        complete: Boolean,
-        chunked: Boolean,
-        contentLength: Option[Long],
-        headers: List[Header.Raw],
-        name: Option[String],
-        start: Int,
-    )
-
-    object ParserState {
-      def initial: ParserState = ParserState(
-        idx = 0,
-        state = false,
-        throwable = None,
-        complete = false,
-        chunked = false,
-        contentLength = None,
-        headers = List.empty,
-        name = None,
-        start = 0,
-      )
-    }
-
-    def parse[F[_]](message: Array[Byte], maxHeaderSize: Int, s: ParserState)(implicit
+    def parseHeaders[F[_]](message: Array[Byte], initIndex: Int, maxHeaderSize: Int)(implicit
         F: MonadThrow[F]
-    ): F[Either[ParserState, HeaderP]] = {
-      var idx: Int = s.idx
-      var state = s.state
-      var throwable: Throwable = s.throwable.orNull
-      var complete = s.complete
-      var chunked: Boolean = s.chunked
-      var contentLength: Option[Long] = s.contentLength
-
-      var headers: List[Header.Raw] = s.headers
-      var name: String = s.name.orNull
-      var start: Int = s.start
+    ): F[Either[Unit, HeaderP]] = {
+      import scala.collection.mutable.ListBuffer
+      var idx = initIndex
+      var state = false
+      var throwable: Throwable = null
+      var complete = false
+      var chunked: Boolean = false
+      var contentLength: Option[Long] = None
+
+      val headers: ListBuffer[Header.Raw] = ListBuffer()
+      var name: String = null
+      var start = initIndex
       val upperBound = Math.min(message.size - 1, maxHeaderSize)
 
       while (!complete && idx <= upperBound) {
@@ -155,7 +127,7 @@ private[ember] object Parser {
               chunked = hValue.contains(chunkedS)
             }
             start = idx + 1 // Next Start is after the CRLF
-            headers = newHeader :: headers // Add Header
+            headers += newHeader // Add Header
             state = false // Go back to Looking for HeaderName or Termination
           }
         }
@@ -165,19 +137,9 @@ private[ember] object Parser {
       if (throwable != null) {
         F.raiseError(ParseHeadersError(throwable))
       } else if (!complete) {
-        ParserState(
-          idx,
-          state,
-          Option(throwable),
-          complete,
-          chunked,
-          contentLength,
-          headers,
-          Option(name),
-          start,
-        ).asLeft.pure[F]
+        ().asLeft.pure[F]
       } else {
-        HeaderP(Headers(headers.reverse), chunked, contentLength, idx).asRight.pure[F]
+        HeaderP(Headers(headers.toList), chunked, contentLength, idx).asRight.pure[F]
       }
     }
 
@@ -197,44 +159,21 @@ private[ember] object Parser {
 
     object ReqPrelude {
 
-      final case class ParserState(
-          idx: Int,
-          state: Byte,
-          complete: Boolean,
-          throwable: Option[Throwable],
-          method: Option[Method],
-          uri: Option[Uri],
-          httpVersion: Option[HttpVersion],
-          start: Int,
-      )
-
-      object ParserState {
-        def initial: ParserState = ParserState(
-          idx = 0,
-          state = 0,
-          complete = false,
-          throwable = None,
-          method = None,
-          httpVersion = None,
-          uri = None,
-          start = 0,
-        )
-      }
-
-      def parse[F[_]](message: Array[Byte], maxHeaderSize: Int, s: ParserState)(implicit
+      // Method SP URI SP HttpVersion CRLF - REST
+      def parsePrelude[F[_]](message: Array[Byte], maxHeaderSize: Int)(implicit
           F: MonadThrow[F]
-      ): F[Either[ParserState, ReqPrelude]] = {
-        var idx = s.idx
-        var state: Byte = s.state
-        var complete = s.complete
-
-        var throwable: Throwable = s.throwable.orNull
-        var method: Method = s.method.orNull
-        var uri: Uri = s.uri.orNull
-        var httpVersion: HttpVersion = s.httpVersion.orNull
-
-        var start = s.start
+      ): F[Either[Unit, ReqPrelude]] = {
+        var idx = 0
+        var state: Byte = 0
+        var complete = false
+
+        var throwable: Throwable = null
+        var method: Method = null
+        var uri: Uri = null
+        var httpVersion: HttpVersion = null
         val upperBound = Math.min(message.size - 1, maxHeaderSize)
+
+        var start = 0
         while (!complete && idx <= upperBound) {
           val value = message(idx)
           (state: @switch) match {
@@ -288,17 +227,7 @@ private[ember] object Parser {
             )
           )
         else if (method == null || uri == null || httpVersion == null)
-          ParserState(
-            idx,
-            state,
-            complete,
-            Option(throwable),
-            Option(method),
-            Option(uri),
-            Option(httpVersion),
-            start,
-          ).asLeft
-            .pure[F]
+          ().asLeft.pure[F]
         else
           ReqPrelude(method, uri, httpVersion, idx).asRight.pure[F]
       }
@@ -320,22 +249,15 @@ private[ember] object Parser {
         read: Read[F],
     )(implicit F: Concurrent[F]): F[(Request[F], Drain[F])] =
       for {
-        x <- MessageP.recurseFind(buffer, read, maxHeaderSize, ReqPrelude.ParserState.initial)(
-          (state, ibuffer) => ReqPrelude.parse(ibuffer, maxHeaderSize, state)
-        )(_.nextIndex)
-        (prelude, buffer2) = x
-        y <- MessageP
-          .recurseFind(
-            buffer2,
-            read,
-            maxHeaderSize,
-            HeaderP.ParserState.initial,
-          )((state, ibuffer) => HeaderP.parse(ibuffer, maxHeaderSize, state))(_.idx)
-          // We've already consumed data to parse the prelude so empty actually means end of stream
-          .adaptError { case _: EmberException.EmptyStream =>
-            EmberException.ReachedEndOfStream()
-          }
-        (headerP, finalBuffer) = y
+        t <- MessageP.recurseFind(buffer, read, maxHeaderSize)(ibuffer =>
+          EitherT(ReqPrelude.parsePrelude[F](ibuffer, maxHeaderSize))
+            .flatMap(prelude =>
+              EitherT(HeaderP.parseHeaders(ibuffer, prelude.nextIndex, maxHeaderSize))
+                .tupleLeft(prelude)
+            )
+            .value
+        ) { case (_, headerP) => headerP.idx }
+        ((prelude, headerP), finalBuffer) = t
 
         baseReq = org.http4s.Request[F](
           method = prelude.method,
@@ -380,23 +302,15 @@ private[ember] object Parser {
           status.responseClass == Status.Informational
 
       for {
-        x <- MessageP.recurseFind(buffer, read, maxHeaderSize, RespPrelude.ParserState.initial)(
-          (state, ibuffer) => RespPrelude.parse(ibuffer, maxHeaderSize, state)
-        )(_.nextIndex)
-        (prelude, buffer2) = x
-        y <- MessageP
-          .recurseFind(
-            buffer2,
-            read,
-            maxHeaderSize,
-            HeaderP.ParserState.initial,
-          )((state, ibuffer) => HeaderP.parse(ibuffer, maxHeaderSize, state))(_.idx)
-          // We've already consumed data to parse the prelude so empty actually means end of stream
-          .adaptError { case _: EmberException.EmptyStream =>
-            EmberException.ReachedEndOfStream()
-          }
-
-        (headerP, finalBuffer) = y
+        t <- MessageP.recurseFind(buffer, read, maxHeaderSize)(ibuffer =>
+          EitherT(RespPrelude.parsePrelude[F](ibuffer, maxHeaderSize))
+            .flatMap(prelude =>
+              EitherT(HeaderP.parseHeaders(ibuffer, prelude.nextIndex, maxHeaderSize))
+                .tupleLeft(prelude)
+            )
+            .value
+        ) { case (_, headerP) => headerP.idx }
+        ((prelude, headerP), finalBuffer) = t
 
         baseResp = org.http4s.Response[F](
           httpVersion = prelude.version,
@@ -434,45 +348,21 @@ private[ember] object Parser {
 
       final case class RespPrelude(version: HttpVersion, status: Status, nextIndex: Int)
 
-      final case class ParserState(
-          idx: Int,
-          state: Byte,
-          complete: Boolean,
-          throwable: Option[Throwable],
-          httpVersion: Option[HttpVersion],
-          codeS: Option[String],
-          status: Option[Status],
-          start: Int,
-      )
-
-      object ParserState {
-        def initial: ParserState = ParserState(
-          idx = 0,
-          state = 0,
-          complete = false,
-          throwable = None,
-          httpVersion = None,
-          codeS = None,
-          status = None,
-          start = 0,
-        )
-      }
-
       // HTTP/1.1 200 OK
       // Status-Line = HTTP-Version SP Status-Code SP Reason-Phrase CRLF
-      def parse[F[_]](buffer: Array[Byte], maxHeaderSize: Int, s: ParserState)(implicit
+      def parsePrelude[F[_]](buffer: Array[Byte], maxHeaderSize: Int)(implicit
           F: MonadThrow[F]
-      ): F[Either[ParserState, RespPrelude]] = {
-        var complete = s.complete
-        var idx = s.idx
-        var throwable: Throwable = s.throwable.orNull
-        var httpVersion: HttpVersion = s.httpVersion.orNull
+      ): F[Either[Unit, RespPrelude]] = {
+        var complete = false
+        var idx = 0
+        var throwable: Throwable = null
+        var httpVersion: HttpVersion = null
 
-        var codeS: String = s.codeS.orNull
+        var codeS: String = null
         // val reason: String = null
-        var status: Status = s.status.orNull
-        var start = s.start
-        var state = s.state // 0 Is for HttpVersion, 1 for Status Code, 2 For Reason Phrase
+        var status: Status = null
+        var start = 0
+        var state = 0 // 0 Is for HttpVersion, 1 for Status Code, 2 For Reason Phrase
         val upperBound = Math.min(buffer.size - 1, maxHeaderSize)
 
         while (!complete && idx <= upperBound) {
@@ -521,17 +411,7 @@ private[ember] object Parser {
         if (throwable != null)
           F.raiseError(RespPreludeError("Encountered Error parsing", Option(throwable)))
         else if (httpVersion == null || status == null)
-          ParserState(
-            idx,
-            state,
-            complete,
-            Option(throwable),
-            Option(httpVersion),
-            Option(codeS),
-            Option(status),
-            start,
-          ).asLeft
-            .pure[F]
+          ().asLeft.pure[F]
         else
           RespPrelude(httpVersion, status, idx).asRight.pure[F]
       }
diff --git a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Client.scala b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Client.scala
index 325ad1bf53..590ba79059 100644
--- a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Client.scala
+++ b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Client.scala
@@ -65,7 +65,7 @@ Connection
     PushPromise - Special Case only after Open or Half-closed(remote)
 
  */
-private[ember] class H2Client[F[_]: Async](
+private[ember] class H2Client[F[_]](
     sg: SocketGroup[F],
     localSettings: H2Frame.Settings.ConnectionSettings,
     tls: TLSContext[F],
@@ -78,7 +78,7 @@ private[ember] class H2Client[F[_]: Async](
         F[org.http4s.Response[F]],
     ) => F[Outcome[F, Throwable, Unit]],
     logger: Logger[F],
-) {
+)(implicit F: Async[F]) {
   import org.http4s._
   import H2Client._
 
@@ -135,24 +135,14 @@ private[ember] class H2Client[F[_]: Async](
     baseSocket <- sg.client(address)
     socket <- {
       if (useTLS) {
+        val tlsParams = Util.mkClientTLSParameters(address.some, enableEndpointValidation)
         for {
           tlsSocket <- tls
             .clientBuilder(baseSocket)
-            .withParameters(
-              H2TLS.transform(Util.mkClientTLSParameters(address.some, enableEndpointValidation))
-            )
+            .withParameters(H2TLS.transform(tlsParams))
             .build
           _ <- Resource.eval(tlsSocket.write(Chunk.empty))
-          protocol <- Resource.eval(H2TLS.protocol(tlsSocket))
-          socketType <- protocol match {
-            case Some("h2") => Resource.pure[F, SocketType](Http2)
-            case Some("http/1.1") => Resource.pure[F, SocketType](Http1)
-            case Some(_) =>
-              Resource.raiseError[F, SocketType, Throwable](
-                new ProtocolException("Unknown protocol")
-              )
-            case None => Resource.pure[F, SocketType](Http1)
-          }
+          socketType <- Resource.eval(parseSocketType(tlsSocket))
         } yield (tlsSocket, socketType)
       } else {
         val socketType = if (priorKnowledge) Http2 else Http1
@@ -162,42 +152,48 @@ private[ember] class H2Client[F[_]: Async](
     }
   } yield socket
 
+  private def parseSocketType(tlsSocket: TLSSocket[F]): F[SocketType] =
+    H2TLS.protocol(tlsSocket).flatMap {
+      case Some("h2") => F.pure(Http2)
+      case Some("http/1.1") => F.pure(Http1)
+      case Some(_) => F.raiseError(new ProtocolException("Unknown protocol"))
+      case None => F.pure(Http1)
+    }
+
   // This Socket Becomes an Http2 Socket
   def fromSocket(
       acc: ByteVector,
       socket: Socket[F],
       key: RequestKey,
-  ): Resource[F, H2Connection[F]] =
-    for {
-      socketAdd <- Resource.eval(RequestKey.getAddress(key))
-      _ <- Resource.eval(socket.write(Chunk.byteVector(Preface.clientBV)))
-      ref <- Resource.eval(Concurrent[F].ref(Map[Int, H2Stream[F]]()))
-      initialWriteBlock <- Resource.eval(Deferred[F, Either[Throwable, Unit]])
-      stateRef <- Resource.eval(
-        Concurrent[F].ref(
-          H2Connection.State(
-            defaultSettings,
-            defaultSettings.initialWindowSize.windowSize,
-            initialWriteBlock,
-            localSettings.initialWindowSize.windowSize,
-            0,
-            0,
-            false,
-            None,
-            None,
+  ): Resource[F, H2Connection[F]] = {
+    def createH2Connection: F[H2Connection[F]] =
+      for {
+        socketAdd <- RequestKey.getAddress(key)
+        _ <- socket.write(Chunk.byteVector(Preface.clientBV))
+        ref <- Concurrent[F].ref(Map[Int, H2Stream[F]]())
+        initialWriteBlock <- Deferred[F, Either[Throwable, Unit]]
+        stateRef <-
+          Concurrent[F].ref(
+            H2Connection.State(
+              defaultSettings,
+              defaultSettings.initialWindowSize.windowSize,
+              initialWriteBlock,
+              localSettings.initialWindowSize.windowSize,
+              0,
+              0,
+              false,
+              None,
+              None,
+            )
           )
-        )
-      )
-      queue <- Resource.eval(cats.effect.std.Queue.unbounded[F, Chunk[H2Frame]]) // TODO revisit
-      hpack <- Resource.eval(Hpack.create[F])
-      settingsAck <- Resource.eval(
-        Deferred[F, Either[Throwable, H2Frame.Settings.ConnectionSettings]]
-      )
-      streamCreationLock <- Resource.eval(cats.effect.std.Semaphore[F](1))
-      // data <- Resource.eval(cats.effect.std.Queue.unbounded[F, Frame.Data])
-      created <- Resource.eval(cats.effect.std.Queue.unbounded[F, Int])
-      closed <- Resource.eval(cats.effect.std.Queue.unbounded[F, Int])
-      h2 = new H2Connection(
+        queue <- cats.effect.std.Queue.unbounded[F, Chunk[H2Frame]] // TODO revisit
+        hpack <- Hpack.create[F]
+        settingsAck <- Deferred[F, Either[Throwable, H2Frame.Settings.ConnectionSettings]]
+        streamCreationLock <- cats.effect.std.Semaphore[F](1)
+        // data <- Resource.eval(cats.effect.std.Queue.unbounded[F, Frame.Data])
+        created <- cats.effect.std.Queue.unbounded[F, Int]
+        closed <- cats.effect.std.Queue.unbounded[F, Int]
+      } yield new H2Connection(
         socketAdd.host,
         socketAdd.port,
         H2Connection.ConnectionType.Client,
@@ -214,61 +210,59 @@ private[ember] class H2Client[F[_]: Async](
         socket,
         logger,
       )
-      _ <- h2.readLoop.background
-      _ <- h2.writeLoop.compile.drain.background
-      _ <-
-        Stream
-          .fromQueueUnterminated(closed)
-          .repeat
-          .evalMap { i =>
-            if (i % 2 != 0) ref.update(m => m - i)
-            else Applicative[F].unit
-          }
-          .compile
-          .drain
-          .background
-      _ <-
-        Stream
-          .fromQueueUnterminated(created)
-          .parEvalMap(10) { i =>
-            val f = if (i % 2 == 0) {
-              val x = for {
-                //
-                stream <- ref.get
-                  .map(_.get(i))
-                  .flatMap(
-                    _.liftTo(new ProtocolException("Stream missing for push promise"))
-                  ) // FOLD
-                // _ <- Sync[F].delay(println(s"Push promise stream acquired for $i"))
-                req <- stream.getRequest
 
-                resp = stream.getResponse.map(
-                  _.covary[F].withBodyStream(stream.readBody)
-                )
-                // _ <- Sync[F].delay(println(s"Push promise request acquired for $i"))
-                outE <- onPushPromise(req, resp).flatMap {
-                  case Outcome.Canceled() => stream.rstStream(H2Error.RefusedStream)
-                  case Outcome.Errored(_) => stream.rstStream(H2Error.RefusedStream)
-                  case Outcome.Succeeded(f) => f
-                }.attempt
-                _ <- ref.update(_ - i)
-                out <- outE.liftTo[F]
-              } yield out
-              x.onError { case e => logger.warn(e)(s"Error Handling Push Promise") }.attempt.void
-            } else Applicative[F].unit
-            f
-          }
-          .compile
-          .drain
-          .onError { case e => logger.info(e)(s"Server Connection Processing Halted") } // Idle etc.
-          .background
+    def clearClosed(h2: H2Connection[F]): F[Unit] =
+      Stream
+        .fromQueueUnterminated(h2.closedStreams)
+        .repeat
+        .foreach(i => if (i % 2 != 0) h2.mapRef.update(m => m - i) else F.unit)
+        .compile
+        .drain
 
-      _ <- Resource.eval(
-        h2.outgoing.offer(
-          Chunk.singleton(H2Frame.Settings.ConnectionSettings.toSettings(localSettings))
-        )
-      )
+    def pullCreatedStreams(h2: H2Connection[F]): F[Unit] = {
+      def processStream(i: Int): F[Unit] =
+        (for {
+          stream <- h2.mapRef.get.flatMap { streamMap =>
+            streamMap.get(i).liftTo(new ProtocolException("Stream missing for push promise"))
+          } // FOLD
+          // _ <- Sync[F].delay(println(s"Push promise stream acquired for $i"))
+          req <- stream.getRequest
+          resp = stream.getResponse.map(_.covary[F].withBodyStream(stream.readBody))
+          // _ <- Sync[F].delay(println(s"Push promise request acquired for $i"))
+          outE <- onPushPromise(req, resp).flatMap {
+            case Outcome.Canceled() => stream.rstStream(H2Error.RefusedStream)
+            case Outcome.Errored(_) => stream.rstStream(H2Error.RefusedStream)
+            case Outcome.Succeeded(f) => f
+          }.attempt
+          _ <- h2.mapRef.update(_ - i)
+          out <- outE.liftTo[F]
+        } yield out)
+          .onError { case e => logger.warn(e)(s"Error Handling Push Promise") }
+          .attempt
+          .void
+
+      Stream
+        .fromQueueUnterminated(h2.createdStreams)
+        .parEvalMap(10)(i => if (i % 2 == 0) processStream(i) else F.unit)
+        .compile
+        .drain
+        .onError { case e => logger.info(e)(s"Server Connection Processing Halted") } // Idle etc.
+    }
+
+    def processSettings(h2: H2Connection[F]): F[Unit] = {
+      val localSetts = H2Frame.Settings.ConnectionSettings.toSettings(localSettings)
+      h2.outgoing.offer(Chunk.singleton(localSetts))
+    }
+
+    for {
+      h2 <- Resource.eval(createH2Connection)
+      _ <- h2.readLoop.background
+      _ <- h2.writeLoop.compile.drain.background
+      _ <- clearClosed(h2).background
+      _ <- pullCreatedStreams(h2).background
+      _ <- Resource.eval(processSettings(h2))
     } yield h2
+  }
 
   def runHttp2Only(req: Request[F], enableEndpointValidation: Boolean): Resource[F, Response[F]] = {
     // Host And Port are required
diff --git a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/PseudoHeaders.scala b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/PseudoHeaders.scala
index cf11159964..35dad8701f 100644
--- a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/PseudoHeaders.scala
+++ b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/PseudoHeaders.scala
@@ -20,7 +20,6 @@ import cats._
 import cats.data._
 import cats.syntax.all._
 import org.http4s._
-import org.typelevel.ci.CIString
 
 /** HTTP/2 pseudo headers */
 private[h2] object PseudoHeaders {
@@ -48,15 +47,13 @@ private[h2] object PseudoHeaders {
       (SCHEME, req.uri.scheme.map(_.value).getOrElse("https"), false) ::
         (PATH, path, false) ::
         (AUTHORITY, req.uri.authority.map(_.toString).getOrElse(""), false) ::
-        req.headers.headers
-          .filterNot(p => connectionHeadersFields.contains(p.name))
-          .map(raw =>
-            (
-              raw.name.toString.toLowerCase(),
-              raw.value,
-              org.http4s.Headers.SensitiveHeaders.contains(raw.name),
-            )
-          ): _*
+        req.headers.headers.map(raw =>
+          (
+            raw.name.toString.toLowerCase(),
+            raw.value,
+            org.http4s.Headers.SensitiveHeaders.contains(raw.name),
+          )
+        ): _*
     )
     l
   }
@@ -126,7 +123,6 @@ private[h2] object PseudoHeaders {
     NonEmptyList(
       (STATUS, response.status.code.toString, false),
       response.headers.headers
-        .filterNot(p => connectionHeadersFields.contains(p.name))
         .map(raw =>
           (
             raw.name.toString.toLowerCase,
@@ -166,15 +162,4 @@ private[h2] object PseudoHeaders {
     }.toOption
       .flatten
 
-  // Connection Specific Headers should be removed when doing h2
-  // This is because connection mechanisms are handled by h2
-  // rather than request/response cycle.
-  // https://httpwg.org/specs/rfc7540.html#rfc.section.8.1.2.2
-  val connectionHeadersFields: Set[CIString] = Set(
-    org.http4s.headers.`Transfer-Encoding`.headerInstance.name,
-    org.http4s.headers.Connection.headerInstance.name,
-    org.http4s.headers.Upgrade.headerInstance.name,
-    org.http4s.headers.`Keep-Alive`.headerInstance.name,
-    CIString("Proxy-Connection"),
-  )
 }
diff --git a/examples/ember/src/main/scala/com/example/http4s/ember/EmberServerH2Example.scala b/examples/ember/src/main/scala/com/example/http4s/ember/EmberServerH2Example.scala
index df6d432fdb..c2789f0bc4 100644
--- a/examples/ember/src/main/scala/com/example/http4s/ember/EmberServerH2Example.scala
+++ b/examples/ember/src/main/scala/com/example/http4s/ember/EmberServerH2Example.scala
@@ -42,10 +42,6 @@ object EmberServerH2Example extends IOApp {
           case _ -> Root / "foo" =>
             // println(req)
             Response[F](Status.Ok).withEntity("Foo Endpoint").pure[F]
-          case _ -> Root / "stream" =>
-            // println("Got stream endpoint")
-            Ok(Stream("Ok").covary[F])
-              .flatTap(Console[F].println)
           case req @ _ -> Root / "trailers" =>
             println(s"Got $req at trailers endpoint")
             req.headers
@@ -84,11 +80,6 @@ object EmberServerH2Example extends IOApp {
         .withHost(ipv4"0.0.0.0")
         .withPort(port"8081")
         .withHttpApp(simpleApp)
-        .withErrorHandler { case error =>
-          Console[F]
-            .println(s"Unexpected error:$error")
-            .as(Response(status = Status.InternalServerError))
-        }
         .build
     } yield ()
 
@@ -100,11 +91,6 @@ object EmberServerH2Example extends IOApp {
         .withHost(ipv4"0.0.0.0")
         .withPort(port"8080")
         .withHttpApp(simpleApp)
-        .withErrorHandler { case error =>
-          Console[F]
-            .println(s"Unexpected error:$error")
-            .as(Response(status = Status.InternalServerError))
-        }
         .build
 
   }
diff --git a/flake.lock b/flake.lock
index a98e073c03..ce6d8c5762 100644
--- a/flake.lock
+++ b/flake.lock
@@ -52,11 +52,11 @@
     },
     "flake-utils_2": {
       "locked": {
-        "lastModified": 1659877975,
-        "narHash": "sha256-zllb8aq3YO3h8B/U0/J1WBgAL8EX5yWf5pMj3G0NAmc=",
+        "lastModified": 1656928814,
+        "narHash": "sha256-RIFfgBuKz6Hp89yRr7+NR5tzIAbn52h8vT6vXkYjZoM=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "c0e246b9b83f637f4681389ecabcb2681b4f3af0",
+        "rev": "7e2a3b3dfd9af950a856d66b0a7d01e3c18aa249",
         "type": "github"
       },
       "original": {
@@ -83,11 +83,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1660494700,
-        "narHash": "sha256-OU9ceb6JvwqUVqFYvYOtpNDwJpHgC3z/dy6GRoaiGgM=",
+        "lastModified": 1659254610,
+        "narHash": "sha256-ee5W5MLWZ3kdx5hwOUs6trOJit+GeTDfG+Lg3rANKoc=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "bc4b4a50c7a105c56f1b712a87818678298deef3",
+        "rev": "67f49b2a3854e8b5e3f9df4422225daa0985f451",
         "type": "github"
       },
       "original": {
@@ -118,11 +118,11 @@
         "nixpkgs": "nixpkgs_2"
       },
       "locked": {
-        "lastModified": 1660579530,
-        "narHash": "sha256-KaIX+bY+M27gNBkt/J2EI98jUlPxxAmsO+1lpI7Qag4=",
+        "lastModified": 1659443857,
+        "narHash": "sha256-iZFgtLpUUiUo0mOVTLe5DzOwPIZylbh8snY0PxJStCc=",
         "owner": "rossabaker",
         "repo": "typelevel-nix",
-        "rev": "323f642c6dfe7d46c1623f429c96cc16ee67bc8f",
+        "rev": "c0796868dd0d4872880041a2799f631fe67be7f7",
         "type": "github"
       },
       "original": {
diff --git a/project/plugins.sbt b/project/plugins.sbt
index 6b52c77d42..fbc9bb4598 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -7,7 +7,7 @@ addSbtPlugin("com.earldouglas" % "xsbt-web-plugin" % "4.2.4")
 addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.11.0")
 addSbtPlugin("com.github.tkawachi" % "sbt-doctest" % "0.10.0")
 addSbtPlugin("org.http4s" % "sbt-http4s-org" % "0.14.4")
-addSbtPlugin("com.github.sbt" % "sbt-native-packager" % "1.9.11")
+addSbtPlugin("com.github.sbt" % "sbt-native-packager" % "1.9.10")
 addSbtPlugin("io.spray" % "sbt-revolver" % "0.9.1")
 addSbtPlugin("pl.project13.scala" % "sbt-jmh" % "0.4.3")
 addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.10.1")


Test patch:
diff --git a/ember-core/shared/src/test/scala/org/http4s/ember/core/ParserSuite.scala b/ember-core/shared/src/test/scala/org/http4s/ember/core/ParserSuite.scala
index 8a5ba94b3a..31546896f0 100644
--- a/ember-core/shared/src/test/scala/org/http4s/ember/core/ParserSuite.scala
+++ b/ember-core/shared/src/test/scala/org/http4s/ember/core/ParserSuite.scala
@@ -45,7 +45,6 @@ class ParsingSuite extends Http4sSuite {
         .emit(s)
         .map(httpifyString)
         .through(fs2.text.utf8.encode[F])
-        .rechunkRandomly(0.1, 0.5)
 
       taking(byteStream).flatMap { read =>
         Parser.Request.parser[F](Int.MaxValue)(Array.emptyByteArray, read).map(_._1)
@@ -57,7 +56,6 @@ class ParsingSuite extends Http4sSuite {
         .emit(s)
         .map(httpifyString)
         .through(fs2.text.utf8.encode[F])
-        .rechunkRandomly(0.1, 0.5)
 
       Resource.eval(
         taking(byteStream).flatMap { read =>
@@ -197,7 +195,6 @@ class ParsingSuite extends Http4sSuite {
           .flatMap { case (resp, _) =>
             resp.body.through(text.utf8.decode).compile.string
           }
-      _ <- IO.println(s"Parsed: $parsed")
     } yield parsed == "{}").assert
   }
 
@@ -393,7 +390,7 @@ class ParsingSuite extends Http4sSuite {
     val asHttp = Helpers.httpifyString(base)
     val bv = asHttp.getBytes()
 
-    Parser.HeaderP.parse[IO](bv, 4096, Parser.HeaderP.ParserState.initial).map {
+    Parser.HeaderP.parseHeaders[IO](bv, 0, 4096).map {
       case Right(headerP) =>
         assertEquals(
           headerP.headers.headers,
@@ -451,15 +448,13 @@ class ParsingSuite extends Http4sSuite {
     val asHttp = Helpers.httpifyString(raw)
     val bv = asHttp.getBytes()
 
-    Parser.Request.ReqPrelude
-      .parse[IO](bv, 4096, Parser.Request.ReqPrelude.ParserState.initial)
-      .map {
-        case Right(prelude) =>
-          assertEquals(prelude.method, Method.GET)
-          assertEquals(prelude.uri, uri"/")
-          assertEquals(prelude.version, HttpVersion.`HTTP/1.1`)
-        case Left(_) => fail("Prelude was not right")
-      }
+    Parser.Request.ReqPrelude.parsePrelude[IO](bv, 4096).map {
+      case Right(prelude) =>
+        assertEquals(prelude.method, Method.GET)
+        assertEquals(prelude.uri, uri"/")
+        assertEquals(prelude.version, HttpVersion.`HTTP/1.1`)
+      case Left(_) => fail("Prelude was not right")
+    }
   }
 
   test("Response Prelude should parse an expected value") {
@@ -468,14 +463,12 @@ class ParsingSuite extends Http4sSuite {
         |""".stripMargin
     val asHttp = Helpers.httpifyString(raw)
     val bv = asHttp.getBytes()
-    Parser.Response.RespPrelude
-      .parse[IO](bv, 4096, Parser.Response.RespPrelude.ParserState.initial)
-      .map {
-        case Right(prelude) =>
-          assertEquals(prelude.version, HttpVersion.`HTTP/1.1`)
-          assertEquals(prelude.status, Status.Ok)
-        case Left(_) => fail("Prelude was not right")
-      }
+    Parser.Response.RespPrelude.parsePrelude[IO](bv, 4096).map {
+      case Right(prelude) =>
+        assertEquals(prelude.version, HttpVersion.`HTTP/1.1`)
+        assertEquals(prelude.status, Status.Ok)
+      case Left(_) => fail("Prelude was not right")
+    }
   }
 
   test("Parser.Response.parser should parse two responses in a row") {
diff --git a/ember-core/shared/src/test/scala/org/http4s/ember/core/StreamingParserSuite.scala b/ember-core/shared/src/test/scala/org/http4s/ember/core/StreamingParserSuite.scala
index f0a954dbc6..6f47c0dd7b 100644
--- a/ember-core/shared/src/test/scala/org/http4s/ember/core/StreamingParserSuite.scala
+++ b/ember-core/shared/src/test/scala/org/http4s/ember/core/StreamingParserSuite.scala
@@ -193,17 +193,6 @@ class StreamingParserSuite extends Http4sSuite {
     }
   }
 
-  // https://github.com/http4s/http4s/issues/6580
-  test("raise end of stream for request with EOF after request line") {
-    val segments = Fixtures.toBytes(List("POST /foo HTTP/1.1\r\n"))
-    (for {
-      read <- Helpers.taking[IO, Byte](List(segments))
-      result <- Parser.Request.parser(Int.MaxValue)(Array.emptyByteArray, read)
-      _ <- result._1.body.compile.drain
-      _ <- result._2
-    } yield ()).intercept[EmberException.ReachedEndOfStream].void
-  }
-
   test("raise end of stream for response") {
     PropF.forAllNoShrinkF(Fixtures.genResponse(min = 3)) { segments =>
       (for {
diff --git a/ember-core/shared/src/test/scala/org/http4s/ember/core/h2/PsuedoHeadersSuite.scala b/ember-core/shared/src/test/scala/org/http4s/ember/core/h2/PsuedoHeadersSuite.scala
deleted file mode 100644
index d190be79f1..0000000000
--- a/ember-core/shared/src/test/scala/org/http4s/ember/core/h2/PsuedoHeadersSuite.scala
+++ /dev/null
@@ -1,59 +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
-package ember.core.h2
-
-import cats.data.NonEmptyList
-import org.http4s.headers._
-import org.typelevel.ci._
-
-class PsuedoHeadersSuite extends Http4sSuite {
-
-  val connectionHeaders: List[Header.ToRaw] = List(
-    Connection.close,
-    `Transfer-Encoding`(NonEmptyList.of(TransferCoding.chunked)),
-    Upgrade(NonEmptyList.of(Protocol(ci"name", None))),
-    `Keep-Alive`.unsafeApply(Some(100L), None, List.empty),
-    "Proxy-Connection" -> "keep-alive",
-  )
-
-  test("requestToHeaders should remove Connection Headers") {
-    val request = Request[fs2.Pure]()
-      .putHeaders(connectionHeaders: _*)
-
-    val test = PseudoHeaders.requestToHeaders(request)
-    val expected = NonEmptyList.of(
-      (":method", "GET", false),
-      (":scheme", "https", false),
-      (":path", "/", false),
-      (":authority", "", false),
-    )
-
-    assertEquals(test, expected)
-
-  }
-  test("responseToHeaders should remove Connection Headers") {
-    val response = Response[fs2.Pure](Status.Ok)
-      .putHeaders(connectionHeaders: _*)
-
-    val test = PseudoHeaders.responseToHeaders(response)
-    val expected = NonEmptyList.of(
-      (":status", "200", false)
-    )
-    assertEquals(test, expected)
-  }
-}
diff --git a/tests/shared/src/test/scala/org/http4s/PathSuite.scala b/tests/shared/src/test/scala/org/http4s/PathSuite.scala
index ddf6233644..26d4628a4c 100644
--- a/tests/shared/src/test/scala/org/http4s/PathSuite.scala
+++ b/tests/shared/src/test/scala/org/http4s/PathSuite.scala
@@ -24,7 +24,22 @@ import org.scalacheck.Prop._
 class PathSuite extends Http4sSuite {
   checkAll("Order[Path]", OrderTests[Path].order)
   checkAll("Semigroup[Path]", SemigroupTests[Path].semigroup)
-  checkAll("Hash[Path]", HashTests[Path].hash)
+
+  test("equals should be consistent with equality") {
+    forAll { (a: Path, b: Path) =>
+      if (a == b)
+        (a.segments == b.segments) && (a.absolute == b.absolute) && (a.endsWithSlash == b.endsWithSlash)
+      else
+        (a.segments != b.segments) || (a.absolute != b.absolute) || (a.endsWithSlash != b.endsWithSlash)
+    }
+  }
+
+  test("hashcode should be consistent with equality") {
+    forAll { (a: Path, b: Path) =>
+      if (a == b) a.## == b.##
+      else a.## != b.##
+    }
+  }
 
   test("merge should be producing a new Path according to rfc3986 5.2.3") {
     forAll { (a: Path, b: Path) =>