Allow re-use of java.net.HttpClient on a per-session basis (#216)

Author: lihaoyi

readme.md

@@ -499,8 +499,18 @@ requests.get(
 
 ## Sessions
 
-A `requests.Session` automatically handles sending/receiving/persisting cookies
-for you across multiple requests:
+`requests.Session` serves two purposes:
+
+- Letting you re-use the underlying HTTP client infrastructure across requests,
+  which improves performance
+
+- Automatically handles sending/receiving/persisting cookies across
+  multiple requests
+
+Note that sessions must be explicitly `close`d after use to prevent leakage 
+of client threads and other resources
+
+### Session Cookies
 
 ```scala
 val s = requests.Session()
@@ -511,6 +521,8 @@ val r2 = s.get("https://httpbin.org/cookies")
 
 r2.text()
 // {"cookies":{"freeform":"test"}}
+
+s.close() // Don't forget to close the session!
 ```
 
 If you want to deal with a website that uses cookies, it's usually easier to use
@@ -537,6 +549,34 @@ r2.text()
 // {"headers":{"X-Special-Header":"omg", ...}}
 ```
 
+In the case of concurrent requests, the last response to be processed will have its
+cookies over-write any previous values.
+
+You can instantiate the session with `persistCookies = false` if you want to avoid
+persisting cookies across requests and purely want `Session`'s for its performance
+or convenience benefits.
+
+### Session Performance
+
+The default `requests.*` API and each `requests.Session` re-uses the same
+underlying `java.net.HttpClient` where possible for performance.
+This saves ~0.2 milliseconds per request, and reduces memory usage from sharing
+the client's threads.
+
+Requests needs to spawn a new client if you customize one of the following keys
+when making a request:
+
+- `proxy`
+- `cert`
+- `sslContext`
+- `verifySslCerts`
+- `connectTimeout`
+
+Whether the performance impact of spawning a new client matters depends on your use
+case. For light scripting making one request here and there, 0.2 milliseconds and a
+few MB of memory overhead is negligible. For applications making large numbers of
+requests, it may be worth instantiated dedicated `requests.Session`
+
 ## Why Requests-Scala?
 
 There is a whole zoo of HTTP clients in the Scala ecosystem. Akka-http, Play-WS,

requests/src/requests/Requester.scala

@@ -16,7 +16,7 @@ import scala.concurrent.{ExecutionException, Future}
 
 import javax.net.ssl.SSLContext
 
-trait BaseSession {
+trait BaseSession extends AutoCloseable {
   def headers: Map[String, String]
   def cookies: mutable.Map[String, HttpCookie]
   def readTimeout: Int
@@ -42,6 +42,21 @@ trait BaseSession {
   lazy val patch = Requester("PATCH", this)
 
   def send(method: String) = Requester(method, this)
+
+  // Executor and HttpClient for this session, lazily initialized
+  lazy val executor: ExecutorService = Executors.newCachedThreadPool()
+  lazy val sharedHttpClient: HttpClient = BaseSession.buildHttpClient(
+    proxy, cert, sslContext, verifySslCerts, connectTimeout, executor
+  )
+
+  /**
+   * Closes the shared HttpClient and its executor. Call this when you're done
+   * with the session to release resources and prevent thread leaks.
+   */
+  def close(): Unit = {
+    BaseSession.closeHttpClient(sharedHttpClient)
+    executor.shutdown()
+  }
 }
 
 object BaseSession {
@@ -50,6 +65,74 @@ object BaseSession {
     "Accept-Encoding" -> "gzip, deflate",
     "Accept" -> "*/*",
   )
+
+  def buildHttpClient(
+    proxy: (String, Int),
+    cert: Cert,
+    sslContext: SSLContext,
+    verifySslCerts: Boolean,
+    connectTimeout: Int,
+    executor: ExecutorService
+  ): HttpClient = {
+    val builder = HttpClient
+      .newBuilder()
+      .executor(executor)
+      .followRedirects(HttpClient.Redirect.NEVER)
+      .proxy(proxy match {
+        case null       => ProxySelector.getDefault
+        case (ip, port) => ProxySelector.of(new InetSocketAddress(ip, port))
+      })
+      .sslContext(
+        if (cert != null)
+          Util.clientCertSSLContext(cert, verifySslCerts)
+        else if (sslContext != null)
+          sslContext
+        else if (!verifySslCerts)
+          Util.noVerifySSLContext
+        else
+          SSLContext.getDefault,
+      )
+      .connectTimeout(Duration.ofMillis(connectTimeout))
+
+    builder.build()
+  }
+
+  /**
+   * Closes an HttpClient, using reflection to handle both Java 21+ (which has close())
+   * and earlier versions (which require accessing internal selector).
+   */
+  def closeHttpClient(httpClient: HttpClient): Unit = {
+    try {
+      val closeMethod = classOf[HttpClient].getMethod("close")
+      closeMethod.invoke(httpClient)
+    } catch {
+      case _: NoSuchMethodException =>
+        // Java < 21: use reflection to access internal selectorManager and close its selector
+        // HttpClient.newBuilder().build() returns HttpClientFacade which wraps HttpClientImpl
+        try {
+          val facadeClass = httpClient.getClass
+          val implField = facadeClass.getDeclaredField("impl")
+          implField.setAccessible(true)
+          val impl = implField.get(httpClient)
+          val selectorManagerField = impl.getClass.getDeclaredField("selmgr")
+          selectorManagerField.setAccessible(true)
+          val selectorManager = selectorManagerField.get(impl)
+          // SelectorManager has a 'selector' field we can close
+          val selectorField = selectorManager.getClass.getDeclaredField("selector")
+          selectorField.setAccessible(true)
+          val selector = selectorField.get(selectorManager)
+          val closeMethod = selector.getClass.getMethod("close")
+          closeMethod.invoke(selector)
+        } catch {
+          case _: Exception =>
+            System.err.println(
+              "requests: Unable to close HttpClient SelectorManager thread. " +
+              "To fix thread leaks on Java <21, add JVM arg: " +
+              "--add-opens java.net.http/jdk.internal.net.http=ALL-UNNAMED"
+            )
+        }
+    }
+  }
 }
 
 object Requester {
@@ -59,6 +142,14 @@ object Requester {
     m.setAccessible(true)
     m
   }
+
+  /** Check if an exception's direct cause is a certificate-related error */
+  def causedByCertificateError(e: Throwable): Boolean = {
+    e.getCause match {
+      case _: java.security.cert.CertificateException | _: java.security.cert.CertPathValidatorException => true
+      case _ => false
+    }
+  }
 }
 
 case class Requester(verb: String, sess: BaseSession) {
@@ -211,28 +302,18 @@ case class Requester(verb: String, sess: BaseSession) {
         new java.net.URL(url + firstSep + encodedParams)
       } else url0
 
-      val executor: ExecutorService = Executors.newSingleThreadExecutor()
+      // Check if we can reuse the session's shared HttpClient
+      val useSharedClient =
+        proxy == sess.proxy &&
+        cert == sess.cert &&
+        sslContext == sess.sslContext &&
+        verifySslCerts == sess.verifySslCerts &&
+        connectTimeout == sess.connectTimeout
+
       val httpClient: HttpClient =
-        HttpClient
-          .newBuilder()
-          .executor(executor)
-          .followRedirects(HttpClient.Redirect.NEVER)
-          .proxy(proxy match {
-            case null       => ProxySelector.getDefault
-            case (ip, port) => ProxySelector.of(new InetSocketAddress(ip, port))
-          })
-          .sslContext(
-            if (cert != null)
-              Util.clientCertSSLContext(cert, verifySslCerts)
-            else if (sslContext != null)
-              sslContext
-            else if (!verifySslCerts)
-              Util.noVerifySSLContext
-            else
-              SSLContext.getDefault,
-          )
-          .connectTimeout(Duration.ofMillis(connectTimeout))
-          .build()
+        if (useSharedClient) sess.sharedHttpClient
+        else BaseSession.buildHttpClient(proxy, cert, sslContext, verifySslCerts, connectTimeout, sess.executor)
+
       try {
 
         val sessionCookieValues = for {
@@ -268,51 +349,42 @@ case class Requester(verb: String, sess: BaseSession) {
         val headersKeyValueAlternating = lastOfEachHeader.values.toList.flatMap {
           case (k, v) => Seq(k, v)
         }
-  
-        val requestBodyInputStream = new PipedInputStream()
-        val requestBodyOutputStream = new PipedOutputStream(requestBodyInputStream)
-  
+
+        // Buffer the request body
+        val requestBodyBuffer = new ByteArrayOutputStream()
+        usingOutputStream(compress.wrap(requestBodyBuffer)) { os => data.write(os) }
+        val requestBodyBytes = requestBodyBuffer.toByteArray
+
         val bodyPublisher: HttpRequest.BodyPublisher =
-          HttpRequest.BodyPublishers.ofInputStream(new Supplier[InputStream] {
-            override def get() = requestBodyInputStream
-          })
-  
+          if (requestBodyBytes.isEmpty) HttpRequest.BodyPublishers.noBody()
+          else HttpRequest.BodyPublishers.ofByteArray(requestBodyBytes)
+
         val requestBuilder =
           HttpRequest
             .newBuilder()
             .uri(url1.toURI)
             .timeout(Duration.ofMillis(readTimeout))
             .headers(headersKeyValueAlternating: _*)
-            .method(
-              upperCaseVerb,
-              (contentLengthHeader.headOption.map(_._2), compress) match {
-                case (Some("0"), _) => HttpRequest.BodyPublishers.noBody()
-                case (Some(n), Compress.None) =>
-                  HttpRequest.BodyPublishers.fromPublisher(bodyPublisher, n.toInt)
-                case _ => bodyPublisher
-              },
-            )
-  
-        val fut = httpClient.sendAsync(
-          requestBuilder.build(),
-          HttpResponse.BodyHandlers.ofInputStream(),
-        )
-  
-        usingOutputStream(compress.wrap(requestBodyOutputStream)) { os => data.write(os) }
-  
+            .method(upperCaseVerb, bodyPublisher)
+
+        def wrapError: PartialFunction[Throwable, Nothing] = {
+          case e: javax.net.ssl.SSLException => throw new InvalidCertException(url, e)
+          case _: HttpConnectTimeoutException | _: HttpTimeoutException =>
+            throw new TimeoutException(url, readTimeout, connectTimeout)
+          case e: java.net.UnknownHostException => throw new UnknownHostException(url, e.getMessage)
+          case e: java.net.ConnectException     => throw new UnknownHostException(url, e.getMessage)
+          case e: IOException if Requester.causedByCertificateError(e) => throw new InvalidCertException(url, e)
+        }
+
         val response =
-          try
-            fut.get()
+          try httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofInputStream())
           catch {
-            case e: ExecutionException =>
-              throw e.getCause match {
-                case e: javax.net.ssl.SSLHandshakeException => new InvalidCertException(url, e)
-                case _: HttpConnectTimeoutException | _: HttpTimeoutException =>
-                  new TimeoutException(url, readTimeout, connectTimeout)
-                case e: java.net.UnknownHostException => new UnknownHostException(url, e.getMessage)
-                case e: java.net.ConnectException     => new UnknownHostException(url, e.getMessage)
-                case e                                => new RequestsException(e.getMessage, Some(e))
-              }
+            case e: Throwable =>
+              wrapError.lift(e)
+                // Sometimes the error we care about is wrapped in an IOException
+                // so check inside to see if there's something we want to handle
+                .orElse(wrapError.lift(e.getCause))
+                .getOrElse(throw new RequestsException(e.getMessage, Some(e)))
           }
 
         val responseCode = response.statusCode()
@@ -337,7 +409,7 @@ case class Requester(verb: String, sess: BaseSession) {
             .flatten
             .exists(_.contains("deflate"))
         def persistCookies() = {
-          if (sess.persistCookies) {
+          if (sess.persistCookies) sess.cookies.synchronized {
             headerFields
               .get("set-cookie")
               .iterator
@@ -439,38 +511,10 @@ case class Requester(verb: String, sess: BaseSession) {
           }
         }
       } finally {
-        // Try to close HttpClient if close() method exists (Java 21+)
-        try {
-          val closeMethod = classOf[HttpClient].getMethod("close")
-          closeMethod.invoke(httpClient)
-        } catch {
-          case _: NoSuchMethodException =>
-            // Java < 21: use reflection to access internal selectorManager and close its selector
-            // HttpClient.newBuilder().build() returns HttpClientFacade which wraps HttpClientImpl
-            try {
-              val facadeClass = httpClient.getClass
-              val implField = facadeClass.getDeclaredField("impl")
-              implField.setAccessible(true)
-              val impl = implField.get(httpClient)
-              val selectorManagerField = impl.getClass.getDeclaredField("selmgr")
-              selectorManagerField.setAccessible(true)
-              val selectorManager = selectorManagerField.get(impl)
-              // SelectorManager has a 'selector' field we can close
-              val selectorField = selectorManager.getClass.getDeclaredField("selector")
-              selectorField.setAccessible(true)
-              val selector = selectorField.get(selectorManager)
-              val closeMethod = selector.getClass.getMethod("close")
-              closeMethod.invoke(selector)
-            } catch {
-              case _: Exception =>
-                System.err.println(
-                  "requests: Unable to close HttpClient SelectorManager thread. " +
-                  "To fix thread leaks on Java <21, add JVM arg: " +
-                  "--add-opens java.net.http/jdk.internal.net.http=ALL-UNNAMED"
-                )
-            }
+        // Only clean up if we created a temporary HttpClient (not using shared)
+        if (!useSharedClient) {
+          BaseSession.closeHttpClient(httpClient)
         }
-        executor.shutdown()
       }
     }
   }