Allow different content types for StreamedBody.RawBinary

Author: sebaciv

rest/jetty/src/main/scala/io/udash/rest/jetty/JettyRestClient.scala

@@ -89,11 +89,9 @@ final class JettyRestClient(
               if (contentLength == -1) {
                 val contentTypeOpt = response.getHeaders.get(HttpHeader.CONTENT_TYPE).opt
                 val mediaTypeOpt = contentTypeOpt.map(MimeTypes.getContentTypeWithoutCharset)
-                val bodyOpt = mediaTypeOpt matchOpt {
-                  case Opt(HttpBody.OctetStreamType) =>
-                    StreamedBody.RawBinary(content = rawContentSubject.doOnSubscriptionCancel(cancelRequest))
-                  case Opt(HttpBody.JsonType) =>
-                    val charset = contentTypeOpt.map(MimeTypes.getCharsetFromContentType).getOrElse(HttpBody.Utf8Charset)
+                val charsetOpt = contentTypeOpt.map(MimeTypes.getCharsetFromContentType)
+                val bodyOpt = (mediaTypeOpt, charsetOpt) matchOpt {
+                  case (Opt(HttpBody.JsonType), Opt(charset)) =>
                     // suboptimal - maybe "online" parsing is possible using Jackson / other lib without waiting for full content ?
                     StreamedBody.JsonList(
                       elements = Observable
@@ -109,6 +107,11 @@ final class JettyRestClient(
                         .onErrorFallbackTo(Observable.raiseError(JettyRestClient.Streaming)),
                       charset = charset,
                     )
+                  case (Opt(mediaType), _) =>
+                    StreamedBody.binary(
+                      content = rawContentSubject.doOnSubscriptionCancel(cancelRequest),
+                      contentType = contentTypeOpt.getOrElse(mediaType),
+                    )
                 }
                 bodyOpt.mapOr(
                   {

rest/src/main/scala/io/udash/rest/raw/StreamedBody.scala

@@ -35,11 +35,10 @@ object StreamedBody extends StreamedBodyLowPrio {
    * The content is delivered as a stream of byte arrays which can be processed incrementally.
    * Useful for large binary files or content that is generated dynamically.
    */
-  final case class RawBinary(content: Observable[Array[Byte]]) extends NonEmpty {
-    val contentType: String = HttpBody.OctetStreamType
-
-    override def toString: String = super.toString
-  }
+  final case class RawBinary(
+    content: Observable[Array[Byte]],
+    override val contentType: String,
+  ) extends NonEmpty
 
   /**
    * Represents a streamed list of JSON values.
@@ -52,8 +51,6 @@ object StreamedBody extends StreamedBodyLowPrio {
     customBatchSize: Opt[Int] = Opt.Empty,
   ) extends NonEmpty {
     val contentType: String = s"${HttpBody.JsonType};charset=$charset"
-
-    override def toString: String = super.toString
   }
 
   /**
@@ -67,6 +64,12 @@ object StreamedBody extends StreamedBodyLowPrio {
 
   def empty: StreamedBody = Empty
 
+  def binary(
+    content: Observable[Array[Byte]],
+    contentType: String = HttpBody.OctetStreamType,
+  ): StreamedBody =
+    RawBinary(content, contentType)
+
   def fromHttpBody(body: HttpBody): StreamedBody = body match {
     case HttpBody.Empty => StreamedBody.Empty
     case nonEmpty: HttpBody.NonEmpty => StreamedBody.Single(nonEmpty)
@@ -89,7 +92,7 @@ object StreamedBody extends StreamedBodyLowPrio {
 
   implicit val rawBinaryBodyForByteArray: AsRawReal[StreamedBody, Observable[Array[Byte]]] =
     AsRawReal.create(
-      bytes => RawBinary(bytes),
+      bytes => StreamedBody.binary(bytes),
       body => StreamedBody.castOrFail[RawBinary](body).content,
     )
 }

rest/src/test/scala/io/udash/rest/raw/RawRestTest.scala

@@ -126,9 +126,9 @@ class RawRestTest extends AnyFunSuite with ScalaFutures with Matchers {
         elements.toListL.map { list =>
           s" application/json;charset=$charset\n[${list.map(_.value).mkString(",")}]"
         }
-      case StreamedBody.RawBinary(content) =>
+      case StreamedBody.RawBinary(content, tpe) =>
         content.toListL.map { list =>
-          s" application/octet-stream\n${list.flatMap(_.iterator.map(b => f"$b%02X")).mkString}"
+          s" $tpe\n${list.flatMap(_.iterator.map(b => f"$b%02X")).mkString}"
         }
       case StreamedBody.Single(body) =>
         Task.now(repr(body, inNewLine = false))