Convert StreamStep to a ListStep for queries and mutations (#174)

core/src/main/scala/caliban/execution/Executor.scala

@@ -27,27 +27,13 @@ object Executor {
     variables: Map[String, InputValue] = Map(),
     fieldWrappers: List[FieldWrapper[R]] = Nil
   ): URIO[R, GraphQLResponse[CalibanError]] = {
+
     val allowParallelism = request.operationType match {
       case OperationType.Query        => true
       case OperationType.Mutation     => false
       case OperationType.Subscription => false
     }
 
-    executePlan(plan, request.field, request.variableDefinitions, variables, allowParallelism, fieldWrappers)
-  }
-
-  private[caliban] def fail(error: CalibanError): UIO[GraphQLResponse[CalibanError]] =
-    IO.succeed(GraphQLResponse(NullValue, List(error)))
-
-  private def executePlan[R](
-    plan: Step[R],
-    root: Field,
-    variableDefinitions: List[VariableDefinition],
-    variableValues: Map[String, InputValue],
-    allowParallelism: Boolean,
-    fieldWrappers: List[FieldWrapper[R]]
-  ): URIO[R, GraphQLResponse[CalibanError]] = {
-
     def reduceStep(
       step: Step[R],
       currentField: Field,
@@ -79,7 +65,7 @@ object Executor {
             case f @ Field(name @ "__typename", _, _, alias, _, _, _) =>
               (alias.getOrElse(name), PureStep(StringValue(objectName)), fieldInfo(f, path))
             case f @ Field(name, _, _, alias, _, _, args) =>
-              val arguments = resolveVariables(args, variableDefinitions, variableValues)
+              val arguments = resolveVariables(args, request.variableDefinitions, variables)
               (
                 alias.getOrElse(name),
                 fields
@@ -94,9 +80,13 @@ object Executor {
             inner.bimap(GenericSchema.effectfulExecutionError(path, _), reduceStep(_, currentField, arguments, path))
           )
         case StreamStep(stream) =>
-          ReducedStep.StreamStep(
-            stream.bimap(GenericSchema.effectfulExecutionError(path, _), reduceStep(_, currentField, arguments, path))
-          )
+          if (request.operationType == OperationType.Subscription) {
+            ReducedStep.StreamStep(
+              stream.bimap(GenericSchema.effectfulExecutionError(path, _), reduceStep(_, currentField, arguments, path))
+            )
+          } else {
+            reduceStep(QueryStep(ZQuery.fromEffect(stream.runCollect.map(ListStep(_)))), currentField, arguments, path)
+          }
       }
 
     def makeQuery(step: ReducedStep[R], errors: Ref[List[CalibanError]]): ZQuery[R, Nothing, ResponseValue] = {
@@ -141,13 +131,16 @@ object Executor {
 
     for {
       errors       <- Ref.make(List.empty[CalibanError])
-      reduced      = reduceStep(plan, root, Map(), Nil)
+      reduced      = reduceStep(plan, request.field, Map(), Nil)
       query        = makeQuery(reduced, errors)
       result       <- query.run
       resultErrors <- errors.get
     } yield GraphQLResponse(result, resultErrors.reverse)
   }
 
+  private[caliban] def fail(error: CalibanError): UIO[GraphQLResponse[CalibanError]] =
+    IO.succeed(GraphQLResponse(NullValue, List(error)))
+
   private def resolveVariables(
     arguments: Map[String, InputValue],
     variableDefinitions: List[VariableDefinition],

core/src/test/scala/caliban/execution/ExecutionSpec.scala

@@ -8,6 +8,7 @@ import caliban.RootResolver
 import caliban.TestUtils._
 import caliban.Value.{ BooleanValue, StringValue }
 import zio.IO
+import zio.stream.ZStream
 import zio.test.Assertion._
 import zio.test._
 
@@ -283,6 +284,34 @@ object ExecutionSpec
             interpreter.execute(query).map(_.errors),
             equalTo(List(ExecutionError("Effect failure", List(Left("a"), Left("b"), Left("c")), Some(e))))
           )
+        },
+        testM("ZStream used in a query") {
+          case class Queries(test: ZStream[Any, Throwable, Int])
+          val interpreter = graphQL(RootResolver(Queries(ZStream(1, 2, 3)))).interpreter
+          val query       = gqldoc("""
+             {
+               test
+             }""")
+
+          assertM(
+            interpreter.execute(query).map(_.data.toString),
+            equalTo("""{"test":[1,2,3]}""")
+          )
+        },
+        testM("ZStream used in a subscription") {
+          case class Queries(test: Int)
+          case class Subscriptions(test: ZStream[Any, Throwable, Int])
+          val interpreter =
+            graphQL(RootResolver(Queries(1), Option.empty[Unit], Subscriptions(ZStream(1, 2, 3)))).interpreter
+          val query = gqldoc("""
+             subscription {
+               test
+             }""")
+
+          assertM(
+            interpreter.execute(query).map(_.data.toString),
+            equalTo("""{"test":<stream>}""")
+          )
         }
       )
     )

vuepress/docs/docs/schema.md

@@ -27,7 +27,7 @@ The table below shows how common Scala types are converted to GraphQL types.
 | ZIO[R, Nothing, A] | A                                                |
 | ZIO[R, E, A]       | Nullable A                                       |
 | Future[A]          | Nullable A                                       |
-| ZStream[R, E, A]   | A                                                |
+| ZStream[R, E, A]   | A (subscription) or List of A (query, mutation)  |
 
 See the [Custom Types](#custom-types) section to find out how to support your own types.
 
@@ -122,6 +122,10 @@ type Queries {
 
 Caliban provides auto-derivation for common types such as `Int`, `String`, `List`, `Option`, etc. but you can also support your own types by providing an implicit instance of `caliban.schema.ArgBuilder`.
 
+::: tip
+There is no `ArgBuilder` for tuples. If you have multiple arguments, use a case class containing all of them instead of a tuple.
+:::
+
 ## Effects
 
 Fields can return ZIO effects. This allows you to leverage all the features provided by ZIO: timeouts, retries, access to ZIO environment, memoizing, etc. An effect will be run every time a query requiring the corresponding field is executed.