Enhance streaming support in REST API: add error handling and update documentation

Author: mrzysztof

guide/guide/.js/src/main/assets/pages/rest.md

@@ -705,7 +705,7 @@ Ultimately, if you don't want to use `Future`s, you may replace it with some oth
 e.g. Monix Task or some IO monad.
 See [supporting result containers other than `Future`](#supporting-result-containers-other-than-future).
 
-See [streaming serialization workflow](#streaming-serialization-workflow) for details on `monix.reactive.Observable` 
+See [streaming serialization workflow](#streaming-serialization-workflow) for details on `monix.reactive.Observable`
 support in streaming REST API methods.
 
 ### Customizing serialization
@@ -949,7 +949,7 @@ computation which yields a `RestResponse` when run.
 
 ### Implementing a server
 
-An existing implementation of REST API trait can be easily turned into a function using 
+An existing implementation of REST API trait can be easily turned into a function using
 `RawRest.asHandleRequest` or `RawRest.asHandleRequestWithStreaming` (for server-side streaming support).
 
 Therefore, the only thing you need to do to expose your REST API trait as an actual web service it to turn
@@ -1096,7 +1096,7 @@ However, this conversion loses the streaming benefits, so it's best used only wh
 
 ### Error Handling
 
-Streaming endpoints handle errors similarly to regular endpoints. When an error occurs during streaming:
+Streaming endpoints handle errors similarly to regular endpoints.
 
 ```scala
 // Server side
@@ -1116,6 +1116,7 @@ client.streamItems("")
 ```
 
 This allows graceful handling of errors that might occur during streaming operations.
+When an error occurs during streaming from the server side, the HTTP connection is closed immediately. This means any in-progress streaming is cut off at the point of error.
 
 ### Advanced Streaming Patterns
 

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

@@ -117,7 +117,6 @@ class RestServlet(
 
   private def writeNonEmptyStreamedBody(
     response: HttpServletResponse,
-    stream: StreamedRestResponse,
     body: StreamedBody.NonEmpty,
   ): Task[Unit] = Task.defer {
     // The Content-Length header is intentionally omitted for streams.
@@ -160,7 +159,11 @@ class RestServlet(
           .map(_ => response.getOutputStream.write("]".getBytes(jsonList.charset)))
     }
   }.onErrorHandle { e =>
-    logger.error(e.getMessage)
+    // When an error occurs during streaming, we immediately close the connection rather than
+    // attempting to send an error response. This is intentional because:
+    // The client has likely already received and started processing partial data
+    // for structured formats (like JSON arrays), the stream is now in an invalid state
+    logger.error("Failure during streaming REST response", e)
     response.getOutputStream.close()
   }
 
@@ -174,7 +177,7 @@ class RestServlet(
       case stream: StreamedRestResponse =>
         stream.body match {
           case StreamedBody.Empty => Task.unit
-          case neBody: StreamedBody.NonEmpty => writeNonEmptyStreamedBody(response, stream, neBody)
+          case neBody: StreamedBody.NonEmpty => writeNonEmptyStreamedBody(response, neBody)
         }
     }
 

rest/src/main/scala/io/udash/rest/RestDataCompanion.scala

@@ -5,9 +5,9 @@ import com.avsystem.commons.meta.MacroInstances
 import com.avsystem.commons.misc.{AbstractValueEnumCompanion, ValueEnum, ValueOf}
 import com.avsystem.commons.rpc.{AsRaw, AsReal}
 import com.avsystem.commons.serialization.{GenCodec, TransparentWrapperCompanion}
-import io.udash.rest.openapi._
+import io.udash.rest.openapi.*
 import io.udash.rest.openapi.RestStructure.NameAndAdjusters
-import io.udash.rest.raw.{HttpBody, JsonValue, PlainValue, RestResponse}
+import io.udash.rest.raw.{HttpBody, JsonValue, PlainValue, RestResponse, StreamedBody, StreamedRestResponse}
 
 trait CodecWithStructure[T] {
   def codec: GenCodec[T]
@@ -94,6 +94,18 @@ abstract class RestDataWrapperCompanion[Wrapped, T](implicit
   implicit def responseAsReal(implicit wrappedAsRaw: AsReal[RestResponse, Wrapped]): AsReal[RestResponse, T] =
     AsReal.fromTransparentWrapping
 
+  implicit def streamedBodyAsRaw(implicit wrappedAsRaw: AsRaw[StreamedBody, Wrapped]): AsRaw[StreamedBody, T] =
+    AsRaw.fromTransparentWrapping
+
+  implicit def streamedBodyAsReal(implicit wrappedAsRaw: AsReal[StreamedBody, Wrapped]): AsReal[StreamedBody, T] =
+    AsReal.fromTransparentWrapping
+
+  implicit def streamedResponseAsRaw(implicit wrappedAsRaw: AsRaw[StreamedRestResponse, Wrapped]): AsRaw[StreamedRestResponse, T] =
+    AsRaw.fromTransparentWrapping
+
+  implicit def streamedResponseAsReal(implicit wrappedAsRaw: AsReal[StreamedRestResponse, Wrapped]): AsReal[StreamedRestResponse, T] =
+    AsReal.fromTransparentWrapping
+
   implicit def restSchema(implicit wrappedSchema: RestSchema[Wrapped]): RestSchema[T] =
     nameAndAdjusters.restSchema(wrappedSchema)