Merge pull request #636 from http4s/http-3-client

Http 3 client

Author: hamnis

build.sbt

@@ -94,6 +94,7 @@ lazy val client = project
       "org.http4s" %% "http4s-client" % http4sVersion,
       "io.netty" % "netty-codec-http2" % netty,
       "io.netty" % "netty-handler-proxy" % netty,
+      "io.netty" % "netty-codec-http3" % netty,
       "org.http4s" %% "http4s-client-testkit" % http4sVersion % Test,
       "org.http4s" %% "http4s-ember-server" % http4sVersion % Test,
       "org.http4s" %% "http4s-dsl" % http4sVersion % Test,

client/src/main/scala/org/http4s/netty/client/EventLoopHolder.scala

@@ -18,36 +18,40 @@ package org.http4s.netty
 package client
 
 import io.netty.bootstrap.Bootstrap
+import io.netty.channel.Channel
 import io.netty.channel.ChannelOption
 import io.netty.channel.MultiThreadIoEventLoopGroup
 import io.netty.channel.MultithreadEventLoopGroup
 import io.netty.channel.epoll.Epoll
+import io.netty.channel.epoll.EpollDatagramChannel
 import io.netty.channel.epoll.EpollIoHandler
 import io.netty.channel.epoll.EpollSocketChannel
 import io.netty.channel.kqueue.KQueue
+import io.netty.channel.kqueue.KQueueDatagramChannel
 import io.netty.channel.kqueue.KQueueIoHandler
 import io.netty.channel.kqueue.KQueueSocketChannel
 import io.netty.channel.nio.NioIoHandler
+import io.netty.channel.socket.DatagramChannel
 import io.netty.channel.socket.SocketChannel
+import io.netty.channel.socket.nio.NioDatagramChannel
 import io.netty.channel.socket.nio.NioSocketChannel
 import io.netty.channel.uring.IoUring
+import io.netty.channel.uring.IoUringDatagramChannel
 import io.netty.channel.uring.IoUringIoHandler
 import io.netty.channel.uring.IoUringSocketChannel
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
-import scala.annotation.nowarn
 import scala.reflect.ClassTag
 
-private[client] final case class EventLoopHolder[A <: SocketChannel](
+private[client] final case class EventLoopHolder[A <: Channel](
     eventLoop: MultithreadEventLoopGroup)(implicit classTag: ClassTag[A]) {
   def runtimeClass: Class[A] = classTag.runtimeClass.asInstanceOf[Class[A]]
 
   def configure(bootstrap: Bootstrap): Bootstrap =
     bootstrap
       .group(eventLoop)
       .channel(runtimeClass)
-      .option(ChannelOption.TCP_NODELAY, java.lang.Boolean.TRUE)
       .option(ChannelOption.SO_KEEPALIVE, java.lang.Boolean.TRUE)
 }
 
@@ -66,7 +70,18 @@ private[client] object EventLoopHolder {
           throw new IllegalStateException("No native transport available"))
     }
 
-  @nowarn("cat=deprecation")
+  def fromUdpTransport(
+      transport: NettyTransport,
+      eventLoopThreads: Int): EventLoopHolder[_ <: DatagramChannel] =
+    transport match {
+      case NettyTransport.Nio =>
+        EventLoopHolder[NioDatagramChannel](
+          new MultiThreadIoEventLoopGroup(eventLoopThreads, NioIoHandler.newFactory()))
+      case n: NettyTransport.Native =>
+        selectUdpTransport(n, eventLoopThreads).getOrElse(
+          throw new IllegalStateException("No native transport available"))
+    }
+
   def selectTransport(
       native: NettyTransport.Native,
       eventLoopThreads: Int): Option[EventLoopHolder[_ <: SocketChannel]] =
@@ -86,7 +101,7 @@ private[client] object EventLoopHolder {
         Some(
           EventLoopHolder[KQueueSocketChannel](
             new MultiThreadIoEventLoopGroup(eventLoopThreads, KQueueIoHandler.newFactory())))
-      case NettyTransport.Auto | NettyTransport.Native =>
+      case NettyTransport.Auto =>
         selectTransport(NettyTransport.IOUring, eventLoopThreads)
           .orElse(selectTransport(NettyTransport.Epoll, eventLoopThreads))
           .orElse(selectTransport(NettyTransport.KQueue, eventLoopThreads))
@@ -98,4 +113,36 @@ private[client] object EventLoopHolder {
           }
       case _ => None
     }
+
+  def selectUdpTransport(
+      native: NettyTransport.Native,
+      eventLoopThreads: Int): Option[EventLoopHolder[_ <: DatagramChannel]] =
+    native match {
+      case NettyTransport.IOUring if IoUring.isAvailable =>
+        logger.info("Using IOUring")
+        Some(
+          EventLoopHolder[IoUringDatagramChannel](
+            new MultiThreadIoEventLoopGroup(eventLoopThreads, IoUringIoHandler.newFactory())))
+      case NettyTransport.Epoll if Epoll.isAvailable =>
+        logger.info("Using Epoll")
+        Some(
+          EventLoopHolder[EpollDatagramChannel](
+            new MultiThreadIoEventLoopGroup(eventLoopThreads, EpollIoHandler.newFactory())))
+      case NettyTransport.KQueue if KQueue.isAvailable =>
+        logger.info("Using KQueue")
+        Some(
+          EventLoopHolder[KQueueDatagramChannel](
+            new MultiThreadIoEventLoopGroup(eventLoopThreads, KQueueIoHandler.newFactory())))
+      case NettyTransport.Auto =>
+        selectUdpTransport(NettyTransport.IOUring, eventLoopThreads)
+          .orElse(selectUdpTransport(NettyTransport.Epoll, eventLoopThreads))
+          .orElse(selectUdpTransport(NettyTransport.KQueue, eventLoopThreads))
+          .orElse {
+            logger.info("Falling back to NIO EventLoopGroup")
+            Some(
+              EventLoopHolder[NioDatagramChannel](
+                new MultiThreadIoEventLoopGroup(eventLoopThreads, NioIoHandler.newFactory())))
+          }
+      case _ => None
+    }
 }

client/src/main/scala/org/http4s/netty/client/Http4sHandler.scala

@@ -32,6 +32,7 @@ import io.netty.handler.codec.http2.Http2SettingsFrame
 import io.netty.handler.timeout.IdleState
 import io.netty.handler.timeout.IdleStateEvent
 import org.http4s._
+import org.http4s.client.RequestKey
 import org.http4s.netty.client.Http4sHandler.Http2GoAwayError
 import org.http4s.netty.client.Http4sHandler.logger
 
@@ -76,7 +77,7 @@ private[netty] class Http4sHandler[F[_]](dispatcher: Dispatcher[F])(implicit F:
 
     val headersFrame = new DefaultHttp2HeadersFrame(
       http2Headers,
-      modelConversion.notAllowedWithBody.contains(request.method))
+      NettyModelConversion.notAllowedWithBody.contains(request.method))
 
     def endOfStream: F[Unit] = request.trailerHeaders.flatMap { headers =>
       val trail =
@@ -86,20 +87,21 @@ private[netty] class Http4sHandler[F[_]](dispatcher: Dispatcher[F])(implicit F:
             HttpConversionUtil.toHttp2Headers(modelConversion.toNettyHeaders(headers), false),
             true)
         }
-      F.delay(writeInEventLoop(trail, channel, key))
+      F.delay(Http4sHandler.writeInEventLoop(trail, channel, key.requestKey)(onException))
     }
 
     val body = if (!headersFrame.isEndStream) {
       (request.body.chunks
         .evalMap(chunk =>
           F.delay(
-            writeInEventLoop(
+            Http4sHandler.writeInEventLoop(
               new DefaultHttp2DataFrame(NettyModelConversion.chunkToBytebuf(chunk), false),
               channel,
-              key))) ++ fs2.Stream.eval(endOfStream)).compile.drain
+              key.requestKey)(onException))) ++ fs2.Stream.eval(endOfStream)).compile.drain
     } else F.unit
 
-    F.delay(writeInEventLoop(headersFrame, channel, key)) >> body
+    F.delay(
+      Http4sHandler.writeInEventLoop(headersFrame, channel, key.requestKey)(onException)) >> body
   }
 
   private[client] def dispatch(
@@ -121,37 +123,12 @@ private[netty] class Http4sHandler[F[_]](dispatcher: Dispatcher[F])(implicit F:
         .evalMap { nettyRequest =>
           F.async[Resource[F, Response[F]]] { cb =>
             promises.enqueue(cb)
-            writeInEventLoop(nettyRequest, channel, key)
+            Http4sHandler.writeInEventLoop(nettyRequest, channel, key.requestKey)(onException)
             F.pure(Some(F.unit))
           }
         }
         .flatMap(identity)
     )
-
-  private def writeInEventLoop(event: AnyRef, channel: Channel, key: Key) =
-    if (channel.eventLoop().inEventLoop) {
-      safedispatch(event, channel, key)
-    } else {
-      channel
-        .eventLoop()
-        .execute(() => safedispatch(event, channel, key))
-    }
-
-  private def safedispatch(event: AnyRef, channel: Channel, key: Key): Unit = void {
-    // always enqueue
-    if (channel.isActive) {
-      logger.trace(s"ch $channel: sending ${event} to $key")
-      // The voidPromise lets us receive failed-write signals from the
-      // exceptionCaught method.
-      channel.writeAndFlush(event, channel.voidPromise)
-      logger.trace(s"ch $channel: after ${event} to $key")
-    } else {
-      // make sure we call all enqueued promises
-      logger.info(s"ch $channel: message dispatched by closed channel to destination $key.")
-      onException(channel, new ClosedChannelException)
-    }
-  }
-
   override def isSharable: Boolean = false
 
   override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = void {
@@ -264,4 +241,30 @@ private object Http4sHandler {
 
   final case class Http2GoAwayError(error: Http2Error, lastStreamId: Int)
       extends Exception(s"http/2 recieved GoAway with ${error.name} and streamId=${lastStreamId}")
+
+  private[client] def writeInEventLoop(event: AnyRef, channel: Channel, key: RequestKey)(
+      onException: (Channel, Throwable) => Unit) =
+    if (channel.eventLoop().inEventLoop) {
+      safedispatch(event, channel, key)(onException)
+    } else {
+      channel
+        .eventLoop()
+        .execute(() => safedispatch(event, channel, key)(onException))
+    }
+
+  private def safedispatch(event: AnyRef, channel: Channel, key: RequestKey)(
+      onException: (Channel, Throwable) => Unit): Unit = void {
+    // always enqueue
+    if (channel.isActive) {
+      logger.trace(s"ch $channel: sending ${event} to $key")
+      // The voidPromise lets us receive failed-write signals from the
+      // exceptionCaught method.
+      channel.writeAndFlush(event, channel.voidPromise)
+      logger.trace(s"ch $channel: after ${event} to $key")
+    } else {
+      // make sure we call all enqueued promises
+      logger.info(s"ch $channel: message dispatched by closed channel to destination $key.")
+      onException(channel, new ClosedChannelException)
+    }
+  }
 }

client/src/main/scala/org/http4s/netty/client/NettyClientBuilder.scala

@@ -20,6 +20,7 @@ package client
 import cats.effect.Async
 import cats.effect.Resource
 import io.netty.bootstrap.Bootstrap
+import io.netty.channel.ChannelOption
 import org.http4s.Headers
 import org.http4s.client.Client
 import org.http4s.headers.`User-Agent`
@@ -122,6 +123,7 @@ class NettyClientBuilder[F[_]](
     Resource.make(F.delay {
       val bootstrap = new Bootstrap()
       EventLoopHolder.fromTransport(transport, eventLoopThreads).configure(bootstrap)
+      bootstrap.option(ChannelOption.TCP_NODELAY, java.lang.Boolean.TRUE)
       nettyChannelOptions.foldLeft(bootstrap) { case (boot, (opt, value)) =>
         boot.option(opt, value)
       }