Fix OpenAPI schema test

Author: sebaciv

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

@@ -24,6 +24,35 @@
         }
       }
     },
+    "/customStreamTask": {
+      "get": {
+        "operationId": "customStreamTask",
+        "parameters": [
+          {
+            "name": "size",
+            "in": "query",
+            "required": true,
+            "explode": false,
+            "schema": {
+              "type": "integer",
+              "format": "int32"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Success",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/DataStream"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/delayedStream": {
       "get": {
         "operationId": "delayedStream",
@@ -67,6 +96,49 @@
         }
       }
     },
+    "/delayedStreamTask": {
+      "get": {
+        "operationId": "delayedStreamTask",
+        "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",
@@ -159,6 +231,13 @@
   ],
   "components": {
     "schemas": {
+      "DataStream": {
+        "type": "array",
+        "items": {
+          "type": "integer",
+          "format": "int32"
+        }
+      },
       "RestEntity": {
         "type": "object",
         "description": "REST entity",

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

@@ -1,26 +1,27 @@
 package io.udash
 package rest
 
-import com.avsystem.commons.rpc.AsRawReal
-import com.avsystem.commons.serialization.json.JsonStringOutput
+import com.avsystem.commons.rpc.{AsRaw, AsRawReal, AsReal}
 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])
+final case class DataStream(source: Observable[Int], metadata: Map[String, String])
+
+object DataStream extends GenCodecRestImplicits {
+  implicit def schema: RestSchema[DataStream] =
+    RestSchema.create(res => RestSchema.seqSchema[Seq, Int].createSchema(res), "DataStream")
 
-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 _ => ???
-      }
+      stream => StreamedBody.JsonList(stream.source.map(AsRaw[JsonValue, Int].asRaw)),
+      rawBody => {
+        val list = StreamedBody.castOrFail[StreamedBody.JsonList](rawBody)
+        DataStream(list.elements.map(AsReal[JsonValue, Int].asReal), Map.empty)
+      },
     )
 }
 
@@ -36,8 +37,8 @@ trait StreamingRestTestApi {
   @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]
 
+  @GET def customStreamTask(@Query size: Int): Task[DataStream]
 }
 object StreamingRestTestApi extends DefaultRestApiCompanion[StreamingRestTestApi] {
 
@@ -64,11 +65,10 @@ object StreamingRestTestApi extends DefaultRestApiCompanion[StreamingRestTestApi
           else throw HttpErrorException.Streaming
         }
 
-    override def delayedStream(size: Int, delayMillis: Long): Observable[Int] = {
+    override def delayedStream(size: Int, delayMillis: Long): Observable[Int] =
       Observable.fromIterable(Range(0, size))
         .zip(Observable.intervalAtFixedRate(delayMillis.millis, delayMillis.millis))
         .map(_._1)
-    }
 
     override def delayedStreamTask(size: Int, delayMillis: Long): Task[Observable[Int]] =
       Task.delay(delayedStream(size, delayMillis))