Implemented CurlMultiSocket

This implementation uses curl multi socket drive, and uses
FileDescriptorPoller from cats effect.
It allows using this library alongside other cats effect libraries.

Author: hnaderi

curl/src/main/scala/org/http4s/curl/http/CurlClient.scala

@@ -19,10 +19,13 @@ package org.http4s.curl.http
 import cats.effect._
 import org.http4s.client.Client
 import org.http4s.curl.unsafe.CurlExecutorScheduler
+import org.http4s.curl.unsafe.CurlMultiSocket
 
 private[curl] object CurlClient {
   def apply(ec: CurlExecutorScheduler): Client[IO] = Client(CurlRequest(ec, _))
 
+  def multiSocket(ms: CurlMultiSocket): Client[IO] = Client(CurlRequest.applyMultiSocket(ms, _))
+
   def get: IO[Client[IO]] = IO.executionContext.flatMap {
     case ec: CurlExecutorScheduler => IO.pure(apply(ec))
     case _ => IO.raiseError(new RuntimeException("Not running on CurlExecutorScheduler"))

curl/src/main/scala/org/http4s/curl/http/CurlRequest.scala

@@ -22,6 +22,7 @@ import org.http4s.Response
 import org.http4s.curl.internal.Utils
 import org.http4s.curl.internal._
 import org.http4s.curl.unsafe.CurlExecutorScheduler
+import org.http4s.curl.unsafe.CurlMultiSocket
 
 private[curl] object CurlRequest {
   private def setup(
@@ -78,6 +79,57 @@ private[curl] object CurlRequest {
       )
     )
 
+  private def setup(
+      handle: CurlEasy,
+      send: RequestSend,
+      recv: RequestRecv,
+      req: Request[IO],
+  ): Resource[IO, Unit] =
+    Utils.newZone.flatMap(implicit zone =>
+      CurlSList().evalMap(headers =>
+        IO {
+          // TODO add in options
+          // handle.setVerbose(true)
+
+          import org.http4s.curl.unsafe.libcurl_const
+          import scala.scalanative.unsafe._
+          import org.http4s.Header
+          import org.http4s.HttpVersion
+          import org.typelevel.ci._
+
+          handle.setCustomRequest(toCString(req.method.renderString))
+
+          handle.setUpload(true)
+
+          handle.setUrl(toCString(req.uri.renderString))
+
+          val httpVersion = req.httpVersion match {
+            case HttpVersion.`HTTP/1.0` => libcurl_const.CURL_HTTP_VERSION_1_0
+            case HttpVersion.`HTTP/1.1` => libcurl_const.CURL_HTTP_VERSION_1_1
+            case HttpVersion.`HTTP/2` => libcurl_const.CURL_HTTP_VERSION_2
+            case HttpVersion.`HTTP/3` => libcurl_const.CURL_HTTP_VERSION_3
+            case _ => libcurl_const.CURL_HTTP_VERSION_NONE
+          }
+          handle.setHttpVersion(httpVersion)
+
+          req.headers // curl adds these headers automatically, so we explicitly disable them
+            .transform(Header.Raw(ci"Expect", "") :: Header.Raw(ci"Transfer-Encoding", "") :: _)
+            .foreach(header => headers.append(header.toString))
+
+          handle.setHttpHeader(headers.toPtr)
+
+          handle.setReadData(Utils.toPtr(send))
+          handle.setReadFunction(RequestSend.readCallback(_, _, _, _))
+
+          handle.setHeaderData(Utils.toPtr(recv))
+          handle.setHeaderFunction(RequestRecv.headerCallback(_, _, _, _))
+
+          handle.setWriteData(Utils.toPtr(recv))
+          handle.setWriteFunction(RequestRecv.writeCallback(_, _, _, _))
+        }
+      )
+    )
+
   def apply(ec: CurlExecutorScheduler, req: Request[IO]): Resource[IO, Response[IO]] = for {
     gc <- GCRoot()
     handle <- CurlEasy()
@@ -89,4 +141,17 @@ private[curl] object CurlRequest {
     _ <- req.body.through(send.pipe).compile.drain.background
     resp <- recv.response()
   } yield resp
+
+  def applyMultiSocket(ms: CurlMultiSocket, req: Request[IO]): Resource[IO, Response[IO]] = for {
+    gc <- GCRoot()
+    handle <- CurlEasy()
+    flow <- FlowControl(handle)
+    send <- RequestSend(flow)
+    recv <- RequestRecv(flow)
+    _ <- gc.add(send, recv)
+    _ <- setup(handle, send, recv, req)
+    _ <- ms.addHandlerTerminating(handle, recv.onTerminated).toResource
+    _ <- req.body.through(send.pipe).compile.drain.background
+    resp <- recv.response()
+  } yield resp
 }

curl/src/main/scala/org/http4s/curl/unsafe/CURLMcode.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2022 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.curl.unsafe
+
+import org.http4s.curl.CurlError
+
+import scala.scalanative.unsafe._
+
+final private[curl] case class CURLMcode(value: CInt) extends AnyVal {
+  @inline def isOk: Boolean = value == 0
+  @inline def isError: Boolean = value != 0
+  @inline def throwOnError: Unit =
+    if (isError) {
+      throw CurlError.fromMCode(this)
+    }
+}

curl/src/main/scala/org/http4s/curl/unsafe/CURLcode.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2022 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.curl.unsafe
+
+import org.http4s.curl.CurlError
+
+import scala.scalanative.unsafe._
+
+final private[curl] case class CURLcode(value: CInt) extends AnyVal {
+  @inline def isOk: Boolean = value == 0
+  @inline def isError: Boolean = value != 0
+  @inline def throwOnError: Unit =
+    if (isError) {
+      throw CurlError.fromCode(this)
+    }
+}

curl/src/main/scala/org/http4s/curl/unsafe/CurlExecutorScheduler.scala

@@ -21,11 +21,13 @@ import cats.effect.kernel.Resource
 import cats.effect.unsafe.PollingExecutorScheduler
 import org.http4s.curl.CurlError
 
+import scala.annotation.nowarn
 import scala.collection.mutable
 import scala.concurrent.duration.Duration
 import scala.scalanative.unsafe._
 import scala.scalanative.unsigned._
 
+@nowarn
 final private[curl] class CurlExecutorScheduler(multiHandle: Ptr[libcurl.CURLM], pollEvery: Int)
     extends PollingExecutorScheduler(pollEvery) {