Repository: http4s/http4s
Pull request #7039 (2023-03-27 18:24:11)

Support ember http/2 over unix sockets
Most of the changes are shoe-horning an `Either[UnixSocketAddress, SocketAddress[Host]]` into the APIs, because we are lacking a universal datatype to represent these. See https://github.com/Comcast/ip4s/issues/466.


Patch:
diff --git a/build.sbt b/build.sbt
index 352bf9e3de..56c4eb8605 100644
--- a/build.sbt
+++ b/build.sbt
@@ -524,6 +524,7 @@ lazy val emberCore = libraryCrossProject("ember-core", CrossType.Full)
       ProblemFilters.exclude[DirectMissingMethodProblem](
         "org.http4s.ember.core.h2.H2Frame#Ping.emptyBV"
       ),
+      ProblemFilters.exclude[Problem]("org.http4s.ember.core.h2.H2Client*"),
     ) ++ {
       if (tlIsScala3.value)
         Seq(
diff --git a/ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala b/ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala
index 46be07730d..29c75d1a15 100644
--- a/ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala
+++ b/ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala
@@ -277,6 +277,7 @@ final class EmberClientBuilder[F[_]: Async: Network] private (
         H2Client.impl[F](
           pushPromiseSupport.getOrElse { case (_, _) => Applicative[F].pure(Outcome.canceled) },
           context,
+          unixSockets,
           logger,
           if (pushPromiseSupport.isDefined) default
           else
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 ff43931d38..b9479dba60 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
@@ -24,6 +24,8 @@ import com.comcast.ip4s._
 import fs2._
 import fs2.io.net._
 import fs2.io.net.tls._
+import fs2.io.net.unixsocket.UnixSocketAddress
+import fs2.io.net.unixsocket.UnixSockets
 import org.http4s.Uri.Authority
 import org.http4s.Uri.Scheme
 import org.http4s._
@@ -67,6 +69,7 @@ Connection
  */
 private[ember] class H2Client[F[_]](
     sg: SocketGroup[F],
+    unix: Option[UnixSockets[F]],
     localSettings: H2Frame.Settings.ConnectionSettings,
     tls: TLSContext[F],
     connections: Ref[
@@ -141,11 +144,21 @@ private[ember] class H2Client[F[_]](
       enableServerNameIndication: Boolean,
   ): Resource[F, (Socket[F], SocketType)] = for {
     address <- Resource.eval(RequestKey.getAddress(key))
-    baseSocket <- sg.client(address)
+    baseSocket <- address match {
+      case Left(address) =>
+        unix
+          .liftTo[Resource[F, *]](
+            new RuntimeException(
+              "No UnixSockets implementation available; use .withUnixSockets(...) to provide one"
+            )
+          )
+          .flatMap(_.client(address))
+      case Right(address) => sg.client(address)
+    }
     socket <- {
       if (useTLS) {
         val tlsParams = Util.mkClientTLSParameters(
-          address.some,
+          address.toOption,
           enableEndpointValidation,
           enableServerNameIndication,
         )
@@ -207,8 +220,7 @@ private[ember] class H2Client[F[_]](
         created <- cats.effect.std.Queue.unbounded[F, Int]
         closed <- cats.effect.std.Queue.unbounded[F, Int]
       } yield new H2Connection(
-        socketAdd.host,
-        socketAdd.port,
+        socketAdd,
         H2Connection.ConnectionType.Client,
         localSettings,
         ref,
@@ -284,6 +296,7 @@ private[ember] class H2Client[F[_]](
   ): Resource[F, Response[F]] = {
     // Host And Port are required
     val key = H2Client.RequestKey.fromRequest(req)
+    val priorKnowledge = req.attributes.contains(H2Keys.Http2PriorKnowledge)
     val useTLS = req.uri.scheme.map(_.value) match {
       case Some("http") => false
       case Some("https") => true
@@ -291,9 +304,8 @@ private[ember] class H2Client[F[_]](
       // this, but with http2, there can be arbitrary schemes
       // but also probably wrong if doing websockets over http/1.1
       case Some(_) => true
-      case None => true
+      case None => !priorKnowledge
     }
-    val priorKnowledge = req.attributes.contains(H2Keys.Http2PriorKnowledge)
     for {
       connection <- Resource.eval(
         getOrCreate(
@@ -326,6 +338,7 @@ private[ember] object H2Client {
           F[org.http4s.Response[F]],
       ) => F[Outcome[F, Throwable, Unit]],
       tlsContext: TLSContext[F],
+      unixSockets: Option[UnixSockets[F]],
       logger: Logger[F],
       settings: H2Frame.Settings.ConnectionSettings = defaultSettings,
       enableEndpointValidation: Boolean,
@@ -355,7 +368,7 @@ private[ember] object H2Client {
         .compile
         .drain
         .background
-      h2 = new H2Client(Network[F], settings, tlsContext, mapH2, onPushPromise, logger)
+      h2 = new H2Client(Network[F], unixSockets, settings, tlsContext, mapH2, onPushPromise, logger)
     } yield (http1Client: TinyClient[F]) => { (req: Request[F]) =>
       val key = H2Client.RequestKey.fromRequest(req)
       val priorKnowledge = req.attributes.contains(H2Keys.Http2PriorKnowledge)
@@ -382,25 +395,35 @@ private[ember] object H2Client {
   case object Http1 extends SocketType
 
   /** Represents a key for requests that can conceivably share a [[Connection]]. */
-  final case class RequestKey(scheme: Scheme, authority: Authority) {
-    override def toString: String = s"${scheme.value}://${authority}"
+  final case class RequestKey(
+      scheme: Scheme,
+      authorityOrAddress: Either[UnixSocketAddress, Authority],
+  ) {
+    override def toString: String =
+      s"${scheme.value}://${authorityOrAddress.fold(_.toString, _.toString)}"
   }
 
   object RequestKey {
     def fromRequest[F[_]](request: Request[F]): RequestKey = {
       val uri = request.uri
-      RequestKey(uri.scheme.getOrElse(Scheme.http), uri.authority.getOrElse(Authority()))
+      val authOrAddr = request.attributes
+        .lookup(Request.Keys.UnixSocketAddress)
+        .toLeft(uri.authority.getOrElse(Authority()))
+      RequestKey(uri.scheme.getOrElse(Scheme.http), authOrAddr)
     }
 
-    def getAddress[F[_]: Sync](requestKey: RequestKey): F[SocketAddress[Host]] =
+    def getAddress[F[_]: Sync](
+        requestKey: RequestKey
+    ): F[Either[UnixSocketAddress, SocketAddress[Host]]] =
       requestKey match {
-        case RequestKey(s, auth) =>
+        case RequestKey(s, Right(auth)) =>
           val port = auth.port.getOrElse(if (s == Uri.Scheme.https) 443 else 80)
           val host = auth.host.value
           for {
             host <- Host.fromString(host).liftTo[F](MissingHost())
             port <- Port.fromInt(port).liftTo[F](MissingPort())
-          } yield SocketAddress[Host](host, port)
+          } yield Right(SocketAddress[Host](host, port))
+        case RequestKey(_, Left(unixAddress)) => Sync[F].pure(Left(unixAddress))
       }
   }
 
diff --git a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Connection.scala b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Connection.scala
index 1d86f228aa..9f14199817 100644
--- a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Connection.scala
+++ b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Connection.scala
@@ -20,15 +20,17 @@ import cats._
 import cats.effect._
 import cats.effect.kernel.Outcome
 import cats.syntax.all._
+import com.comcast.ip4s.Host
+import com.comcast.ip4s.SocketAddress
 import fs2._
 import fs2.io.net.Socket
 import fs2.io.net.SocketException
+import fs2.io.net.unixsocket.UnixSocketAddress
 import org.typelevel.log4cats.Logger
 import scodec.bits._
 
 private[h2] class H2Connection[F[_]](
-    host: com.comcast.ip4s.Host,
-    port: com.comcast.ip4s.Port,
+    address: Either[UnixSocketAddress, SocketAddress[Host]],
     connectionType: H2Connection.ConnectionType,
     localSettings: H2Frame.Settings.ConnectionSettings,
     val mapRef: Ref[F, Map[Int, H2Stream[F]]],
@@ -46,6 +48,8 @@ private[h2] class H2Connection[F[_]](
     logger: Logger[F],
 )(implicit F: Temporal[F]) {
 
+  private[this] def addrStr = address.fold(_.toString, _.toString)
+
   def initiateLocalStream: F[H2Stream[F]] = for {
     t <- state.modify { s =>
       val highestIsEven = s.highestStream % 2 == 0
@@ -160,7 +164,7 @@ private[h2] class H2Connection[F[_]](
             state.update(s => s.copy(writeWindow = s.writeWindow - fullDataSize)) >>
               socket.isOpen.ifM(
                 socket.write(Chunk.byteVector(bv)) >>
-                  chunk.traverse_(frame => logger.debug(s"$host:$port Write - $frame")),
+                  chunk.traverse_(frame => logger.debug(s"$addrStr Write - $frame")),
                 new SocketException("Socket closed when attempting to write").raiseError,
               )
           } else {
@@ -174,7 +178,7 @@ private[h2] class H2Connection[F[_]](
             }
             socket.isOpen.ifM(
               socket.write(Chunk.byteVector(bv)) >>
-                nonData.traverse_(frame => logger.debug(s"$host:$port Write - $frame")),
+                nonData.traverse_(frame => logger.debug(s"$addrStr Write - $frame")),
               new SocketException("Socket closed when attempting to write").raiseError,
             ) >>
               s.writeBlock.get.rethrow >>
@@ -193,7 +197,7 @@ private[h2] class H2Connection[F[_]](
   //  Currently will backpressure at the data frame till its cleared
 
   def readLoop: F[Unit] = {
-    def connectionTerminated: String = s"Connection $host:$port readLoop Terminated"
+    def connectionTerminated: String = s"Connection $addrStr readLoop Terminated"
     val readFromSocket: F[Option[Chunk[Byte]]] =
       socket.read(localSettings.initialWindowSize.windowSize)
 
@@ -522,7 +526,7 @@ private[h2] class H2Connection[F[_]](
     def readLoopAux(acc: ByteVector): F[Unit] =
       readNextFrame(acc).flatMap {
         case Some((frame, nacc)) =>
-          logger.debug(s"$host:$port Read - $frame") >>
+          logger.debug(s"$addrStr Read - $frame") >>
             state.get.flatMap(processFrame(frame, _)) >>
             readLoopAux(nacc)
         case None => F.unit
diff --git a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Server.scala b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Server.scala
index 4cb1dadfb1..e733b551f0 100644
--- a/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Server.scala
+++ b/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Server.scala
@@ -24,6 +24,7 @@ import cats.syntax.all._
 import fs2._
 import fs2.io.IOException
 import fs2.io.net._
+import fs2.io.net.unixsocket.UnixSocketAddress
 import org.http4s._
 import org.typelevel.ci._
 import org.typelevel.log4cats.Logger
@@ -194,8 +195,10 @@ private[ember] object H2Server {
       F.sleep(1.seconds) >> stateRef.get.map(_.closed).ifM(F.unit, holdWhileOpen(stateRef))
 
     def initH2Connection: F[H2Connection[F]] = for {
-      address <- socket.remoteAddress
-      (remotehost, remoteport) = (address.host, address.port)
+      address <- socket.remoteAddress.attempt.map(
+        // TODO, only used for logging
+        _.leftMap(_ => UnixSocketAddress("unknown.sock"))
+      )
       ref <- Concurrent[F].ref(Map[Int, H2Stream[F]]())
       initialWriteBlock <- Deferred[F, Either[Throwable, Unit]]
       stateRef <-
@@ -220,8 +223,7 @@ private[ember] object H2Server {
       created <- cats.effect.std.Queue.unbounded[F, Int]
       closed <- cats.effect.std.Queue.unbounded[F, Int]
     } yield new H2Connection(
-      remotehost,
-      remoteport,
+      address,
       H2Connection.ConnectionType.Server,
       localSettings,
       ref,


Test patch:
diff --git a/ember-server/js-jvm/src/test/scala/org/http4s/ember/server/EmberUnixSocketSuite.scala b/ember-server/js-jvm/src/test/scala/org/http4s/ember/server/EmberUnixSocketSuite.scala
index 76830a1c1e..104d6ecf5e 100644
--- a/ember-server/js-jvm/src/test/scala/org/http4s/ember/server/EmberUnixSocketSuite.scala
+++ b/ember-server/js-jvm/src/test/scala/org/http4s/ember/server/EmberUnixSocketSuite.scala
@@ -24,35 +24,56 @@ import fs2.io.net.unixsocket.UnixSockets
 import org.http4s._
 import org.http4s.client.middleware.UnixSocket
 import org.http4s.ember.client.EmberClientBuilder
+import org.http4s.ember.core.h2.H2Keys
 
 import scala.concurrent.duration._
 
 class EmberUnixSocketSuite extends Http4sSuite {
 
-  test("Should be able to connect to a server created") {
-    def app = HttpApp.liftF(Response[IO](Status.Ok).withEntity("Hello Unix Sockets!").pure[IO])
-    val test = for {
-      path <- Files[IO].tempFile(None, "", "sock", None)
-      _ <- Resource.eval(Files[IO].deleteIfExists(path))
-      localSocket = UnixSocketAddress(path.toString)
-      _ <- EmberServerBuilder
-        .default[IO]
-        .withUnixSocketConfig(UnixSockets[IO], localSocket)
-        .withHttpApp(app)
-        .build
-      client <- EmberClientBuilder
-        .default[IO]
-        .withUnixSockets(UnixSockets[IO])
-        .build
-        .map(UnixSocket(localSocket))
-      _ <- Resource.eval(IO.sleep(4.seconds))
-      resp <- client.run(Request[IO](Method.GET))
-      body <- Resource.eval(resp.bodyText.compile.string)
-    } yield {
-      assertEquals(resp.status, Status.Ok)
-      assertEquals(body, "Hello Unix Sockets!")
+  def run(
+      setupServer: EmberServerBuilder[IO] => EmberServerBuilder[IO],
+      setupClient: EmberClientBuilder[IO] => EmberClientBuilder[IO],
+      setupRequest: Request[IO] => Request[IO],
+  ): IO[Unit] = {
+
+    val msg = "Hello Unix Sockets!"
+
+    def app = HttpApp.liftF(Response[IO](Status.Ok).withEntity(msg).pure[IO])
+
+    Files[IO].tempFile(None, "", "sock", None).use { path =>
+      val localSocket = UnixSocketAddress(path.toString)
+
+      val server = setupServer(
+        EmberServerBuilder
+          .default[IO]
+          .withUnixSocketConfig(UnixSockets[IO], localSocket)
+          .withShutdownTimeout(1.second)
+          .withHttpApp(app)
+      ).build
+
+      val client = setupClient(
+        EmberClientBuilder
+          .default[IO]
+          .withUnixSockets(UnixSockets[IO])
+      ).build.map(UnixSocket(localSocket))
+
+      val request = setupRequest(Request[IO](Method.GET))
+
+      Files[IO].deleteIfExists(path) *>
+        (server *> client).use { client =>
+          IO.sleep(4.seconds) *>
+            client.expect[String](request).assertEquals(msg) *>
+            client.expect[String](request).assertEquals(msg)
+        }
     }
-    test.use(_ => ().pure[IO])
+  }
+
+  test("http/1.1") {
+    run(identity(_), identity(_), identity(_))
+  }
+
+  test("http/2") {
+    run(_.withHttp2, _.withHttp2, _.withAttribute(H2Keys.Http2PriorKnowledge, ()))
   }
 
 }