Handle streaming vs. chunked

Author: nafg

requests/src/requests/Model.scala

@@ -77,6 +77,7 @@ trait RequestBlob{
 object RequestBlob{
   object EmptyRequestBlob extends RequestBlob{
     def write(out: java.io.OutputStream): Unit = ()
+    override def contentLength = Some(0)
   }
 
   implicit class ByteSourceRequestBlob[T](x: T)(implicit f: T => geny.Writable) extends RequestBlob{

requests/src/requests/Requester.scala

@@ -1,15 +1,15 @@
 package requests
 
-import java.io.*
-import java.net.http.*
-import java.net.{UnknownHostException as _, *}
+import java.io._
+import java.net.http._
+import java.net.{UnknownHostException => _, _}
 import java.nio.ByteBuffer
 import java.time.Duration
 import java.util.concurrent.Flow
 import java.util.function.Supplier
 import java.util.zip.{GZIPInputStream, InflaterInputStream}
 
-import scala.collection.JavaConverters.*
+import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.concurrent.{ExecutionException, Future}
 
@@ -200,26 +200,6 @@ case class Requester(verb: String,
           .connectTimeout(Duration.ofMillis(connectTimeout))
           .build()
 
-      val requestBodyInputStream = new PipedInputStream()
-
-      val requestBuilder =
-        HttpRequest.newBuilder()
-          .uri(url1.toURI)
-          .method(upperCaseVerb, HttpRequest.BodyPublishers.ofInputStream(new Supplier[InputStream] {
-            override def get() = requestBodyInputStream
-          }))
-          .timeout(Duration.ofMillis(readTimeout))
-
-      for ((k, v) <- blobHeaders) requestBuilder.header(k, v)
-
-      for ((k, v) <- sess.headers) requestBuilder.header(k, v)
-
-      for ((k, v) <- headers) requestBuilder.header(k, v)
-
-      for ((k, v) <- compress.headers) requestBuilder.header(k, v)
-
-      auth.header.foreach(requestBuilder.header("Authorization", _))
-
       val sessionCookieValues = for {
         c <- (sess.cookies ++ cookies).valuesIterator
         if !c.hasExpired
@@ -228,20 +208,44 @@ case class Requester(verb: String,
       } yield (c.getName, c.getValue)
 
       val allCookies = sessionCookieValues ++ cookieValues
-      if (allCookies.nonEmpty) {
-        requestBuilder.header(
-          "Cookie",
-          allCookies
+
+      val allHeaders =
+        blobHeaders ++
+          sess.headers ++
+          headers ++
+          compress.headers ++
+          auth.header.map("Authorization" -> _) ++
+          (if (allCookies.isEmpty) None
+          else Some("Cookie" -> allCookies
             .map { case (k, v) => s"""$k="$v"""" }
             .mkString("; ")
-        )
-      }
+          ))
+      val allHeadersFlat = allHeaders.toList.flatMap { case (k, v) => Seq(k, v) }
+
+      val requestBodyInputStream = new PipedInputStream()
+      val requestBodyOutputStream = new PipedOutputStream(requestBodyInputStream)
+
+      val bodyPublisher: HttpRequest.BodyPublisher =
+        HttpRequest.BodyPublishers.ofInputStream(new Supplier[InputStream] {
+          override def get() = requestBodyInputStream
+        })
 
-      val request = requestBuilder.build()
+      val requestBuilder =
+        HttpRequest.newBuilder()
+          .uri(url1.toURI)
+          .timeout(Duration.ofMillis(readTimeout))
+          .headers(allHeadersFlat: _*)
+          .method(upperCaseVerb,
+            data.contentLength match {
+              case Some(0)                              => HttpRequest.BodyPublishers.noBody()
+              case Some(n) if compress == Compress.None => HttpRequest.BodyPublishers.fromPublisher(bodyPublisher, n)
+              case _                                    => bodyPublisher
+            }
+          )
 
-      val fut = httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofInputStream())
+      val fut = httpClient.sendAsync(requestBuilder.build(), HttpResponse.BodyHandlers.ofInputStream())
 
-      usingOutputStream(compress.wrap(new PipedOutputStream(requestBodyInputStream))) { os =>
+      usingOutputStream(compress.wrap(requestBodyOutputStream)) { os =>
         data.write(os)
       }