Add unit tests for advanced streaming types and non-streaming implementation

Author: mrzysztof

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

@@ -1172,9 +1172,6 @@ object DataStream {
 }
 
 trait CustomStreamingApi {
-  /** Returns a custom streaming type */
-  def getDataStream(query: String): DataStream[SearchResult]
-
   /** Returns a Task that produces a custom streaming type */
   def prepareAndStreamData(id: String): Task[DataStream[DataPoint]]
 }

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.{HttpErrorException, RawRest}
+import io.udash.rest.raw.RawRest
 import io.udash.testing.AsyncUdashSharedTest
 import monix.eval.Task
 import monix.execution.Scheduler
@@ -194,4 +194,40 @@ trait StreamingRestApiTestScenarios extends RestApiTest {
       }
     }
   }
+
+  "streaming with non-streaming client" in {
+    val standardProxy = RawRest.fromHandleRequest[StreamingRestTestApi](clientHandle)
+    standardProxy.simpleStream(3).toListL.materialize.runToFuture.map {
+      case Failure(exception: UnsupportedOperationException) =>
+        assert(exception.getMessage == "Streaming unsupported by the client")
+      case Failure(otherException) =>
+        fail(s"Expected UnsupportedOperationException but got ${otherException.getClass.getName}: ${otherException.getMessage}")
+      case Success(_) =>
+        fail("Expected UnsupportedOperationException but operation succeeded")
+    }
+  }
+
+  "task of observable stream" in {
+    val testTask = for {
+      proxyResults <- streamingProxy.delayedStreamTask(3, 50).flatMap(_.toListL)
+      implResults <- streamingImpl.delayedStreamTask(3, 50).flatMap(_.toListL)
+    } yield {
+      assert(proxyResults.map(mkDeep) == implResults.map(mkDeep))
+    }
+    testTask.runToFuture
+  }
+
+  "custom stream task" in {
+    val testTask = for {
+      proxyResults <- streamingProxy.customStreamTask(3)
+      implResults <- streamingImpl.customStreamTask(3)
+      proxyObs <- proxyResults.source.toListL
+      implObs <- implResults.source.toListL
+    } yield {
+      assert(proxyResults.metadata == implResults.metadata)
+      assert(proxyObs == implObs)
+    }
+    testTask.runToFuture
+  }
+
 }

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

@@ -1,11 +1,29 @@
 package io.udash
 package rest
 
-import io.udash.rest.raw.HttpErrorException
+import com.avsystem.commons.rpc.AsRawReal
+import com.avsystem.commons.serialization.json.JsonStringOutput
+import io.udash.rest.openapi.RestSchema
+import io.udash.rest.raw.{HttpErrorException, JsonValue, StreamedBody}
+import monix.eval.Task
 import monix.reactive.Observable
 
 import scala.concurrent.duration.*
 
+case class DataStream(source: Observable[Int], metadata: Map[String, String])
+
+object DataStream {
+  implicit def schema: RestSchema[DataStream] = ???
+  implicit def dataStreamAsRawReal: AsRawReal[StreamedBody, DataStream] =
+    AsRawReal.create(
+      stream => StreamedBody.JsonList(stream.source.map(i => JsonValue(JsonStringOutput.write(i)))),
+      {
+        case StreamedBody.JsonList(e, c) => DataStream(e.map(_.value.toInt), Map.empty)
+        case _ => ???
+      }
+    )
+}
+
 trait StreamingRestTestApi {
   @GET def simpleStream(size: Int): Observable[Int]
 
@@ -16,6 +34,10 @@ trait StreamingRestTestApi {
   @POST def errorStream(@Query immediate: Boolean): Observable[RestEntity]
 
   @GET def delayedStream(@Query size: Int, @Query delayMillis: Long): Observable[Int]
+
+  @GET def delayedStreamTask(@Query size: Int, @Query delayMillis: Long): Task[Observable[Int]]
+  @GET def customStreamTask(@Query size: Int): Task[DataStream]
+
 }
 object StreamingRestTestApi extends DefaultRestApiCompanion[StreamingRestTestApi] {
 
@@ -47,5 +69,15 @@ object StreamingRestTestApi extends DefaultRestApiCompanion[StreamingRestTestApi
         .zip(Observable.intervalAtFixedRate(delayMillis.millis, delayMillis.millis))
         .map(_._1)
     }
+
+    override def delayedStreamTask(size: Int, delayMillis: Long): Task[Observable[Int]] =
+      Task.delay(delayedStream(size, delayMillis))
+
+    override def customStreamTask(size: Int): Task[DataStream] = Task {
+      DataStream(
+        Observable.fromIterable(Range(0, size)),
+        Map.empty
+      )
+    }
   }
 }