fix emissions of streaming messages halting run

Author: martypitt

preflight-core/build.gradle.kts

@@ -26,7 +26,7 @@ tasks.register("publishAllMavenLocal") {
 
 allprojects {
     group = "com.orbitalhq.preflight"
-    version = "0.1.0-M1"
+    version = "0.1.0-M2"
 
     repositories {
         mavenCentral()

preflight-core/preflight-runtime/src/main/kotlin/com/orbitalhq/preflight/dsl/MarkdownSpec.kt

@@ -1,11 +1,16 @@
 package com.orbitalhq.preflight.dsl
 
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.orbitalhq.models.Provided
+import com.orbitalhq.models.TypedInstance
+import com.orbitalhq.models.json.right
 import com.orbitalhq.preflight.spec.StubMode
 import com.orbitalhq.preflight.spec.TestSpec
 import com.orbitalhq.preflight.spec.TestSpecReader
+import com.orbitalhq.schemas.fqn
 import com.orbitalhq.stubbing.StubService
 import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.flow.flowOf
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
@@ -48,9 +53,9 @@ private fun OrbitalSpec.registerSpec(
                     }
                 }
                 StubMode.STREAM -> {
-                    val emitter = stubService.addResponseEmitter(stub.operationName)
-                    stub.messages?.forEach { message ->
-                        emitter.next(message)
+                    stubService.addResponseFlow(stub.operationName) { _,_ ->
+                        val messages = messagesAsTypedInstanceResponses(stub, stubService.schema!!)
+                        flowOf(*messages.toTypedArray())
                     }
                 }
             }

preflight-core/preflight-runtime/src/main/kotlin/com/orbitalhq/preflight/dsl/StubHelper.kt

@@ -1,5 +1,16 @@
 package com.orbitalhq.preflight.dsl
 
+import arrow.core.Either
+import com.orbitalhq.models.Provided
+import com.orbitalhq.models.TypedInstance
+import com.orbitalhq.models.json.right
+import com.orbitalhq.preflight.spec.Stub
+import com.orbitalhq.query.StreamErrorMessage
+import com.orbitalhq.schemas.Schema
+import com.orbitalhq.schemas.fqn
+import kotlin.collections.map
+import kotlin.collections.orEmpty
+
 /**
  * Generates a Stub response builder for an operation with
  * the provided name.
@@ -20,4 +31,14 @@ data class StubResponseBuilder(val operationName: String) {
     }
 }
 
-data class StubScenario(val operationName: String, val response: String)
+data class StubScenario(val operationName: String, val response: String)
+
+fun messagesAsTypedInstanceResponses(stub: Stub, schema: Schema): List<Either<StreamErrorMessage, TypedInstance>> {
+    val (_,operation) = schema.remoteOperation(stub.operationName.fqn())
+    val streamType = operation.returnType.typeParameters[0]
+    val messagesAsTypedInstances = stub.messages.orEmpty().map { json ->
+        TypedInstance.from(streamType, json, source = Provided, schema = schema)
+            .right()
+    }
+    return messagesAsTypedInstances
+}