Add streaming support testing to REST API with new endpoints and error handling

Author: mrzysztof

rest/.jvm/src/test/resources/StreamingRestTestApi.json

@@ -24,6 +24,49 @@
         }
       }
     },
+    "/delayedStream": {
+      "get": {
+        "operationId": "delayedStream",
+        "parameters": [
+          {
+            "name": "size",
+            "in": "query",
+            "required": true,
+            "explode": false,
+            "schema": {
+              "type": "integer",
+              "format": "int32"
+            }
+          },
+          {
+            "name": "delayMillis",
+            "in": "query",
+            "required": true,
+            "explode": false,
+            "schema": {
+              "type": "integer",
+              "format": "int64"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Success",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "type": "array",
+                  "items": {
+                    "type": "integer",
+                    "format": "int32"
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/errorStream": {
       "post": {
         "operationId": "errorStream",

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

@@ -100,7 +100,7 @@ import scala.concurrent.duration.*
                           Observable
                             .fromIterator(Task.eval(input.readList().iterator(_.asInstanceOf[JsonStringInput].readRawJson())))
                             .map(JsonValue(_))
-                        },
+                        }.onErrorFallbackTo(Observable.raiseError(HttpErrorException.Streaming)),
                       charset = charset,
                     )
                 }

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

@@ -150,4 +150,7 @@ abstract class RestServerOpenApiImplCompanion[Implicits, Real](protected val imp
 
   final def asHandleRequest(real: Real): RawRest.HandleRequest =
     RawRest.asHandleRequest(real)
+
+  final def asHandleRequestWithStreaming(real: Real): RawRest.HandleRequestWithStreaming =
+    RawRest.asHandleRequestWithStreaming(real)
 }

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

@@ -2,9 +2,9 @@ package io.udash
 package rest
 package raw
 
-import com.avsystem.commons.meta._
+import com.avsystem.commons.meta.*
 import com.avsystem.commons.misc.{AbstractValueEnum, AbstractValueEnumCompanion, EnumCtx}
-import com.avsystem.commons.rpc._
+import com.avsystem.commons.rpc.*
 
 import scala.util.control.NoStackTrace
 
@@ -56,6 +56,7 @@ case class HttpErrorException(code: Int, payload: HttpBody = HttpBody.Empty, cau
 object HttpErrorException {
   def plain(code: Int, message: String, cause: Throwable = null): HttpErrorException =
     HttpErrorException(code, HttpBody.plain(message), cause)
+  val Streaming: HttpErrorException = plain(400, "HTTP stream failure")
 }
 
 final case class RestRequest(method: HttpMethod, parameters: RestParameters, body: HttpBody) {

rest/src/test/scala/io/udash/rest/RestApiTest.scala

@@ -4,7 +4,7 @@ package rest
 import cats.implicits.catsSyntaxTuple2Semigroupal
 import com.avsystem.commons.*
 import com.avsystem.commons.misc.ScalaDurationExtensions.durationIntOps
-import io.udash.rest.raw.RawRest
+import io.udash.rest.raw.{HttpErrorException, RawRest}
 import io.udash.testing.AsyncUdashSharedTest
 import monix.eval.Task
 import monix.execution.Scheduler
@@ -13,6 +13,7 @@ import org.scalactic.source.Position
 import org.scalatest.time.{Millis, Seconds, Span}
 import org.scalatest.{Assertion, BeforeAndAfterEach}
 
+import scala.concurrent.TimeoutException
 import scala.concurrent.duration.FiniteDuration
 
 abstract class RestApiTest extends AsyncUdashSharedTest with BeforeAndAfterEach {
@@ -28,7 +29,7 @@ abstract class RestApiTest extends AsyncUdashSharedTest with BeforeAndAfterEach
 
   override protected def beforeEach(): Unit = {
     super.beforeEach()
-    impl.resetCounter()
+    impl.resetCounter() // Reset non-streaming counter
   }
 
   final val serverHandle: RawRest.HandleRequest =
@@ -168,8 +169,29 @@ trait StreamingRestApiTestScenarios extends RestApiTest {
     testStream(_.errorStream(immediate = true))
   }
 
-  // TODO streaming - does not work on client side
-  "mid-stream error" ignore {
+  "mid-stream error" in {
     testStream(_.errorStream(immediate = false))
   }
+
+  "slow source stream" in {
+    testStream(_.delayedStream(size = 3, delayMillis = 100))
+  }
+
+  "client-side timeout on slow stream" in {
+    val streamTask = streamingProxy
+      .delayedStream(size = 10, delayMillis = 200)
+      .toListL
+
+    val timeoutTask = streamTask.timeout(500.millis).materialize
+
+    timeoutTask.runToFuture.map { result =>
+      assert(result.isFailure, "Stream should have failed due to timeout")
+      result match {
+        case Failure(ex) =>
+          assert(ex.isInstanceOf[TimeoutException], s"Expected TimeoutException, but got $ex")
+          succeed
+        case Success(_) => fail("Stream succeeded unexpectedly despite timeout")
+      }
+    }
+  }
 }

rest/src/test/scala/io/udash/rest/SomeServerApiImpl.scala

@@ -1,11 +1,31 @@
 package io.udash.rest
 
+import monix.reactive.Observable
+
 import scala.concurrent.Future
 
 final class SomeServerApiImpl {
   @GET
   def thingy(param: Int): Future[String] = Future.successful((param - 1).toString)
 
+  @GET
+  def streamingNumbers(count: Int): Observable[Int] =
+    Observable.fromIterable(1 to count)
+
+  @POST
+  def streamEcho(values: List[Int]): Observable[Int] =
+    Observable.fromIterable(values)
+
+  @GET
+  def streamBinary(chunkSize: Int): Observable[Array[Byte]] = {
+    val content = "HelloWorld".getBytes
+    Observable.fromIterable(content.grouped(chunkSize).toSeq)
+  }
+
+  @GET
+  def streamEmpty(): Observable[Array[Byte]] =
+    Observable.empty
+
   val subapi = new SomeServerSubApiImpl
 }
 object SomeServerApiImpl extends DefaultRestServerApiImplCompanion[SomeServerApiImpl]

rest/src/test/scala/io/udash/rest/StreamingRestTestApi.scala

@@ -2,10 +2,11 @@ package io.udash
 package rest
 
 import io.udash.rest.raw.HttpErrorException
-import monix.execution.Scheduler
+import monix.eval.Task
 import monix.reactive.Observable
 
-import scala.concurrent.duration._
+import java.util.concurrent.atomic.AtomicBoolean
+import scala.concurrent.duration.*
 
 trait StreamingRestTestApi {
   @GET def simpleStream(size: Int): Observable[Int]
@@ -15,11 +16,11 @@ trait StreamingRestTestApi {
   @POST def binaryStream(): Observable[Array[Byte]]
 
   @POST def errorStream(@Query immediate: Boolean): Observable[RestEntity]
+
+  @GET def delayedStream(@Query size: Int, @Query delayMillis: Long): Observable[Int]
 }
 object StreamingRestTestApi extends DefaultRestApiCompanion[StreamingRestTestApi] {
 
-  import Scheduler.Implicits.global
-
   final class Impl extends StreamingRestTestApi {
 
     override def simpleStream(size: Int): Observable[Int] =
@@ -40,7 +41,14 @@ object StreamingRestTestApi extends DefaultRestApiCompanion[StreamingRestTestApi
       else
         Observable.fromIterable(Range(0, 3)).map { i =>
           if (i < 2) RestEntity(RestEntityId(i.toString), "first")
-          else throw HttpErrorException.plain(400, "later bad")
-        }
+          else throw HttpErrorException.Streaming
+          }
+
+
+    override def delayedStream(size: Int, delayMillis: Long): Observable[Int] = {
+      Observable.fromIterable(Range(0, size))
+        .zip(Observable.intervalAtFixedRate(delayMillis.millis, delayMillis.millis))
+        .map(_._1)
+    }
   }
-}
+}