Enhance documentation for streaming support

Author: Krzysztof Maliszewski

rest/.jvm/src/main/scala/io/udash/rest/RestServlet.scala

@@ -111,18 +111,20 @@ class RestServlet(
     stream: StreamedRestResponse,
     body: StreamedBody.NonEmpty,
   ): Task[Unit] = Task.defer {
+    // The Content-Length header is intentionally omitted for streams.
+    // This signals to the client that the response body size is not predetermined and will be streamed.
+    // Clients implementing the streaming part of the REST interface contract MUST be prepared
+    // to handle responses without Content-Length by reading data incrementally until the stream completes.
     body match {
       case single: StreamedBody.Single =>
         Task.eval(writeNonEmptyBody(response, single.body))
       case binary: StreamedBody.RawBinary =>
-        // TODO streaming document no content length behaviour in relation to the client
         response.setContentType(binary.contentType)
         binary.content.bufferTumbling(stream.batchSize).consumeWith(Consumer.foreach { batch =>
           batch.foreach(e => response.getOutputStream.write(e))
           response.getOutputStream.flush()
         })
       case jsonList: StreamedBody.JsonList =>
-        // TODO streaming document no content length behaviour in relation to the client
         response.setContentType(jsonList.contentType)
         jsonList.elements
           .bufferTumbling(stream.batchSize)

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

@@ -20,8 +20,18 @@ import java.nio.charset.Charset
 import scala.concurrent.CancellationException
 import scala.concurrent.duration.*
 
-/** TODO streaming doc */
-final class JettyRestClient(
+/**
+ * A REST client implementation based on the Eclipse Jetty HTTP client library.
+ * Supports both standard request/response interactions and handling of streamed responses.
+ *
+ * Streaming responses allow processing large amounts of data without buffering the entire
+ * response body in memory. This client activates streaming mode *only* when the server's
+ * response headers *do not* include a `Content-Length`.
+ *
+ * @param client The configured Jetty `HttpClient` instance.
+ * @param defaultMaxResponseLength Default maximum size (in bytes) for buffering non-streamed responses.
+ * @param defaultTimeout Default timeout for requests.
+ */final class JettyRestClient(
   client: HttpClient,
   defaultMaxResponseLength: Int = JettyRestClient.DefaultMaxResponseLength,
   defaultTimeout: Duration = JettyRestClient.DefaultTimeout,
@@ -37,7 +47,16 @@ final class JettyRestClient(
       asHandleRequestWithStreaming(baseUri, customMaxResponseLength, customTimeout)
     )
 
-  /** TODO streaming doc */
+  /**
+   * Creates a request handler with streaming support that can be used to make REST calls.
+   * The handler supports both regular responses and streaming responses, allowing for
+   * incremental processing of large payloads through Observable streams.
+   *
+   * @param baseUrl Base URL for the REST service
+   * @param customMaxResponseLength Optional maximum response length override for non-streamed responses
+   * @param customTimeout Optional timeout override
+   * @return A handler that can process REST requests with streaming capabilities
+   */
   def asHandleRequestWithStreaming(
     baseUrl: String,
     customMaxResponseLength: OptArg[Int] = OptArg.Empty,
@@ -59,7 +78,9 @@ final class JettyRestClient(
 
             override def onHeaders(response: Response): Unit = {
               super.onHeaders(response)
-              // TODO streaming document content length behaviour
+              // When Content-Length is not provided (-1), process the response as a stream
+              // since we can't determine the full size in advance. This enables handling
+              // chunked transfer encoding and streaming responses.
               val contentLength = response.getHeaders.getLongField(HttpHeader.CONTENT_LENGTH)
               if (contentLength == -1) {
                 val contentTypeOpt = response.getHeaders.get(HttpHeader.CONTENT_TYPE).opt
@@ -122,7 +143,8 @@ final class JettyRestClient(
                 val httpResp = result.getResponse
                 val contentLength = httpResp.getHeaders.getLongField(HttpHeader.CONTENT_LENGTH)
                 if (contentLength != -1) {
-                  // TODO streaming client-side handle errors ?
+                  // For responses with known content length, we handle them as regular (non-streamed) responses
+                  // Any errors will be propagated through the callback's Failure channel
                   val restResponse = StreamedRestResponse(
                     code = httpResp.getStatus,
                     headers = parseHeaders(httpResp),
@@ -142,7 +164,15 @@ final class JettyRestClient(
       }
   }
 
-  /** TODO streaming doc */
+  /**
+   * Creates a `RawRest.HandleRequest` which handles standard REST requests by buffering the entire response.
+   * This does *not* support streaming responses.
+   *
+   * @param baseUrl The base URL for the REST service.
+   * @param customMaxResponseLength Optional override for the maximum response length.
+   * @param customTimeout Optional override for the request timeout.
+   * @return A `RawRest.HandleRequest` that buffers responses.
+   */
   def asHandleRequest(
     baseUrl: String,
     customMaxResponseLength: OptArg[Int] = OptArg.Empty,

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

@@ -141,7 +141,11 @@ trait RawRest {
   def asHandleRequestWithStreaming(metadata: RestMetadata[_]): HandleRequestWithStreaming =
     RawRest.resolveAndHandle(metadata)(handleResolvedWithStreaming)
 
-  // TODO doc for compatibility
+  /**
+   * Handles a resolved REST call and returns a standard RestResponse.
+   * This method is maintained for backward compatibility with non-streaming clients.
+   * It delegates to handleResolvedWithStreaming and converts any streaming response to a standard response.
+   */
   def handleResolved(request: RestRequest, resolved: ResolvedCall): Task[RestResponse] =
     StreamedRestResponse.fallbackToRestResponse(handleResolvedWithStreaming(request, resolved))
 

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

@@ -1,26 +1,25 @@
 package io.udash
 package rest.raw
 
-import com.avsystem.commons._
+import com.avsystem.commons.*
 import com.avsystem.commons.misc.ImplicitNotFound
 import com.avsystem.commons.rpc.{AsRaw, AsReal}
 import io.udash.rest.raw.RawRest.FromTask
-import io.udash.rest.raw.StreamedBody.castOrFail
 import io.udash.rest.util.Utils
 import monix.eval.{Task, TaskLike}
 import monix.reactive.Observable
 
 import scala.annotation.implicitNotFound
 
-/** TODO streaming doc */
+/** Base trait for REST response types, either standard or streaming. Contains common properties like status code and headers. */
 sealed trait AbstractRestResponse {
   def code: Int
   def headers: IMapping[PlainValue]
 
   final def isSuccess: Boolean = code >= 200 && code < 300
 }
 
-/** TODO streaming doc */
+/** Standard REST response containing a status code, headers, and a body. The body is loaded fully in memory as an HttpBody. */
 final case class RestResponse(
   code: Int,
   headers: IMapping[PlainValue],
@@ -112,7 +111,10 @@ trait RestResponseLowPrio { this: RestResponse.type =>
   ): ImplicitNotFound[AsRaw[RestResponse, T]] = ImplicitNotFound()
 }
 
-/** TODO streaming doc */
+/**
+ * Streaming REST response containing a status code, headers, and a streamed body.
+ * Unlike standard RestResponse, the body content can be delivered incrementally through a reactive stream.
+ */
 final case class StreamedRestResponse(
   code: Int,
   headers: IMapping[PlainValue],
@@ -129,7 +131,10 @@ final case class StreamedRestResponse(
 
 object StreamedRestResponse extends StreamedRestResponseLowPrio {
 
-  /** TODO streaming doc */
+  /**
+   * Converts a StreamedRestResponse to a standard RestResponse by materializing streamed content.
+   * This is useful for compatibility with APIs that don't support streaming.
+   */
   def fallbackToRestResponse(response: StreamedRestResponse): Task[RestResponse] = {
     val httpBody: Task[HttpBody] = response.body match {
       case StreamedBody.Empty =>
@@ -152,7 +157,10 @@ object StreamedRestResponse extends StreamedRestResponseLowPrio {
     httpBody.map(RestResponse(response.code, response.headers, _))
   }
 
-  /** TODO doc */
+  /**
+   * Converts any AbstractRestResponse to a standard RestResponse by materializing streamed content if necessary.
+   * This is useful for compatibility with APIs that don't support streaming.
+   */
   def fallbackToRestResponse(response: Task[AbstractRestResponse]): Task[RestResponse] =
     response.flatMap {
       case restResponse: RestResponse => Task.now(restResponse)

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

@@ -31,14 +31,22 @@ object StreamedBody extends StreamedBodyLowPrio {
     def contentType: String
   }
 
-  /** TODO streaming doc */
+  /**
+   * Represents a binary streamed response body.
+   * 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
   }
 
-  /** TODO streaming doc */
+  /**
+   * Represents a streamed list of JSON values.
+   * Each element in the stream is a complete JSON value, allowing for incremental processing
+   * of potentially large collections without loading everything into memory at once.
+   */
   final case class JsonList(
     elements: Observable[JsonValue],
     charset: String = HttpBody.Utf8Charset,
@@ -48,7 +56,11 @@ object StreamedBody extends StreamedBodyLowPrio {
     override def toString: String = super.toString
   }
 
-  /** TODO streaming doc */
+  /**
+   * Represents a single non-empty HTTP body that will be delivered as a streaming response.
+   * Used when the content is already fully loaded but needs to be returned through a streaming API
+   * for consistency with other streaming operations.
+   */
   final case class Single(body: HttpBody.NonEmpty) extends NonEmpty {
     override def contentType: String = body.contentType
   }