add minimal support for resolve function returning IO

.github/workflows/ci.yml

@@ -64,7 +64,7 @@ jobs:
         run: sbt ++${{ matrix.scala }} test
 
       - name: Compress target directories
-        run: tar cf targets.tar modules/core/target modules/parser/target modules/test-monix/target modules/benchmarks/target modules/derivation/target modules/test-fs2/target modules/ast/target modules/test-cats-effect/target target modules/sangria/target project/target
+        run: tar cf targets.tar modules/core/target modules/cats-effect-experimental/target modules/parser/target modules/test-monix/target modules/benchmarks/target modules/derivation/target modules/test-fs2/target modules/ast/target target modules/sangria/target project/target
 
       - name: Upload target directories
         uses: actions/upload-artifact@v2

build.sbt

@@ -48,7 +48,7 @@ lazy val root = project
     derivation,
     sangriaTestMonix,
     sangriaTestFS2,
-    sangriaTestCatsEffect,
+    sangriaCatsEffectExperimental,
     sangria)
   .settings(inThisBuild(projectInfo))
   .settings(
@@ -284,16 +284,16 @@ lazy val sangriaTestFS2 = project
   )
   .disablePlugins(MimaPlugin)
 
-lazy val sangriaTestCatsEffect = project
-  .in(file("modules/test-cats-effect"))
-  .withId("sangria-test-cats-effect")
+lazy val sangriaCatsEffectExperimental = project
+  .in(file("modules/cats-effect-experimental"))
+  .withId("sangria-cats-effect-experimental")
   .dependsOn(core % "compile->compile;test->test")
-  .settings(scalacSettings ++ shellSettings ++ noPublishSettings)
+  .settings(scalacSettings ++ shellSettings)
   .settings(
-    name := "sangria-test-cats-effect",
-    description := "Tests with Cats Effect",
+    name := "sangria-cats-effect-experimental",
+    description := "Experimental support for Cats Effect",
     libraryDependencies ++= List(
-      "org.typelevel" %% "cats-effect" % "3.4.8" % Test,
+      "org.typelevel" %% "cats-effect" % "3.4.8",
       "org.sangria-graphql" %% "sangria-circe" % "1.3.2" % Test
     )
   )

modules/cats-effect-experimental/src/main/scala/sangria/catseffect/execution/IOExecutionScheme.scala

@@ -0,0 +1,22 @@
+package sangria.catseffect.execution
+
+import cats.effect.IO
+import cats.effect.unsafe.implicits.global
+import sangria.execution.{AsyncExecutionScheme, AsyncToFuture}
+
+import scala.concurrent.{ExecutionContext, Future}
+
+/** Prepare an [[sangria.execution.ExecutionScheme]] for [[IO]]. If you want to use another effect,
+  * use the same bricks to build your own.
+  */
+object IOExecutionScheme {
+  // sangria is using an implicit ExecutionContext at different places.
+  // For the moment, we need to expose one.
+  implicit val ec: ExecutionContext = global.compute
+
+  // ideally we would need only this.
+  implicit val asyncExecutionScheme: AsyncExecutionScheme[IO] =
+    new AsyncExecutionScheme[IO](new AsyncToFuture[IO] {
+      override def toFuture[A](f: IO[A]): Future[A] = f.unsafeToFuture()
+    })
+}

modules/cats-effect-experimental/src/main/scala/sangria/catseffect/schema/AsyncValue.scala

@@ -0,0 +1,18 @@
+package sangria.catseffect.schema
+
+import cats.effect.Async
+import sangria.schema.LeafAction
+
+import scala.concurrent.ExecutionContext
+import scala.language.implicitConversions
+
+case class AsyncValue[Ctx, Val, F[_]: Async](value: F[Val]) extends LeafAction[Ctx, Val] {
+  override def map[NewVal](fn: Val => NewVal)(implicit
+      ec: ExecutionContext): AsyncValue[Ctx, NewVal, F] =
+    new AsyncValue(Async[F].map(value)(fn))
+}
+
+object AsyncValue {
+  implicit def asyncAction[Ctx, Val, F[_]: Async](value: F[Val]): LeafAction[Ctx, Val] =
+    AsyncValue(value)
+}

modules/cats-effect-experimental/src/main/scala/sangria/execution/AsyncExecutionScheme.scala

@@ -0,0 +1,31 @@
+package sangria.execution
+
+import cats.effect.Async
+
+import scala.concurrent.{ExecutionContext, Future}
+
+/** An [[ExecutionScheme]] that is capable of using [[sangria.catseffect.schema.AsyncValue]].
+  *
+  * Its result is an [[Async]].
+  */
+class AsyncExecutionScheme[F[_]: Async](
+    val asyncToFuture: AsyncToFuture[F]
+) extends ExecutionScheme {
+  private val asyncF: Async[F] = Async[F]
+
+  override type Result[Ctx, Res] = F[Res]
+
+  override def failed[Ctx, Res](error: Throwable): Result[Ctx, Res] = asyncF.raiseError(error)
+
+  override def onComplete[Ctx, Res](result: Result[Ctx, Res])(op: => Unit)(implicit
+      ec: ExecutionContext): Result[Ctx, Res] =
+    asyncF.flatMap(result)(r => asyncF.map(asyncF.delay(op))(_ => r))
+
+  override def flatMapFuture[Ctx, Res, T](future: Future[T])(resultFn: T => Result[Ctx, Res])(
+      implicit ec: ExecutionContext): Result[Ctx, Res] =
+    asyncF.flatMap(asyncF.fromFuture(asyncF.delay(future)))(resultFn)
+
+  override val resolverBuilder: ResolverBuilder = new AsyncResolverBuilder[F](asyncToFuture)
+
+  override def extended: Boolean = false
+}

modules/cats-effect-experimental/src/main/scala/sangria/execution/AsyncResolver.scala

@@ -0,0 +1,169 @@
+package sangria.execution
+
+import cats.effect.Async
+import sangria.ast
+import sangria.ast.{Document, SourceMapper}
+import sangria.catseffect.schema.AsyncValue
+import sangria.execution.deferred.DeferredResolver
+import sangria.marshalling.ResultMarshaller
+import sangria.schema._
+
+import scala.concurrent.{ExecutionContext, Future}
+
+/** The [[AsyncResolver]] is using the [[FutureResolver]] under the hood. So we need a way to
+  * transform our [[Async]] into a [[Future]] for now.
+  */
+trait AsyncToFuture[F[_]] {
+  def toFuture[A](f: F[A]): Future[A]
+}
+
+private[execution] class AsyncResolverBuilder[F[_]: Async](asyncToFuture: AsyncToFuture[F])
+    extends ResolverBuilder {
+  override def build[Ctx](
+      marshaller: ResultMarshaller,
+      middlewareCtx: MiddlewareQueryContext[Ctx, _, _],
+      schema: Schema[Ctx, _],
+      valueCollector: ValueCollector[Ctx, _],
+      variables: Map[String, VariableValue],
+      fieldCollector: FieldCollector[Ctx, _],
+      userContext: Ctx,
+      exceptionHandler: ExceptionHandler,
+      deferredResolver: DeferredResolver[Ctx],
+      sourceMapper: Option[SourceMapper],
+      deprecationTracker: DeprecationTracker,
+      middleware: List[(Any, Middleware[Ctx])],
+      maxQueryDepth: Option[Int],
+      deferredResolverState: Any,
+      preserveOriginalErrors: Boolean,
+      validationTiming: TimeMeasurement,
+      queryReducerTiming: TimeMeasurement,
+      queryAst: Document)(implicit executionContext: ExecutionContext): Resolver[Ctx] =
+    new AsyncResolver[Ctx, F](
+      marshaller,
+      middlewareCtx,
+      schema,
+      valueCollector,
+      variables,
+      fieldCollector,
+      userContext,
+      exceptionHandler,
+      deferredResolver,
+      sourceMapper,
+      deprecationTracker,
+      middleware,
+      maxQueryDepth,
+      deferredResolverState,
+      preserveOriginalErrors,
+      validationTiming,
+      queryReducerTiming,
+      queryAst,
+      asyncToFuture
+    )
+}
+
+/** The [[Resolver]] that is used to resolve [[AsyncValue]].
+  *
+  * For now, it's using the [[FutureResolver]] under the hood. Later, we can update its
+  * implementation to avoid using any [[Future]].
+  */
+private[execution] class AsyncResolver[Ctx, F[_]: Async](
+    val marshaller: ResultMarshaller,
+    middlewareCtx: MiddlewareQueryContext[Ctx, _, _],
+    schema: Schema[Ctx, _],
+    valueCollector: ValueCollector[Ctx, _],
+    variables: Map[String, VariableValue],
+    fieldCollector: FieldCollector[Ctx, _],
+    userContext: Ctx,
+    exceptionHandler: ExceptionHandler,
+    deferredResolver: DeferredResolver[Ctx],
+    sourceMapper: Option[SourceMapper],
+    deprecationTracker: DeprecationTracker,
+    middleware: List[(Any, Middleware[Ctx])],
+    maxQueryDepth: Option[Int],
+    deferredResolverState: Any,
+    preserveOriginalErrors: Boolean,
+    validationTiming: TimeMeasurement,
+    queryReducerTiming: TimeMeasurement,
+    queryAst: ast.Document,
+    asyncToFuture: AsyncToFuture[F]
+)(implicit executionContext: ExecutionContext)
+    extends Resolver[Ctx] {
+
+  private val asyncF: Async[F] = Async[F]
+
+  private val delegate = new FutureResolver[Ctx](
+    marshaller,
+    middlewareCtx,
+    schema,
+    valueCollector,
+    variables,
+    fieldCollector,
+    userContext,
+    exceptionHandler,
+    deferredResolver,
+    sourceMapper,
+    deprecationTracker,
+    middleware,
+    maxQueryDepth,
+    deferredResolverState,
+    preserveOriginalErrors,
+    validationTiming,
+    queryReducerTiming,
+    queryAst
+  ) { del =>
+    override protected def resolveLeafAction(
+        path: ExecutionPath,
+        tpe: ObjectType[Ctx, _],
+        userCtx: Ctx,
+        astFields: Vector[ast.Field],
+        field: Field[Ctx, _],
+        updateCtx: Option[MappedCtxUpdate[Ctx, Any, Any]])(
+        action: LeafAction[Ctx, Any]): (ast.Field, Resolve) =
+      action match {
+        case a: AsyncValue[Ctx, Any, F] =>
+          val f = asyncToFuture.toFuture[Any](a.value)
+          super.resolveFutureValue(path, tpe, userCtx, astFields, field, updateCtx)(FutureValue(f))
+
+        case action: StandardLeafAction[Ctx, Any] =>
+          super.resolveStandardLeafAction(path, tpe, userCtx, astFields, field, updateCtx)(action)
+
+        case other => unresolvableLeafAction(path, tpe, astFields, updateCtx)(other)
+      }
+
+    override protected def handleScheme(
+        result: Future[((Vector[RegisteredError], del.marshaller.Node), Ctx)],
+        scheme: ExecutionScheme): scheme.Result[Ctx, del.marshaller.Node] =
+      scheme match {
+        case s: AsyncExecutionScheme[F] =>
+          val r: F[((Vector[RegisteredError], del.marshaller.Node), Ctx)] =
+            asyncF.fromFuture(asyncF.delay(result))
+          handleSchemeF(r, s).asInstanceOf[scheme.Result[Ctx, del.marshaller.Node]]
+
+        case _ =>
+          super.handleScheme(result, scheme)
+      }
+
+    private def handleSchemeF(
+        result: F[((Vector[RegisteredError], del.marshaller.Node), Ctx)],
+        scheme: AsyncExecutionScheme[F]): scheme.Result[Ctx, del.marshaller.Node] =
+      asyncF.map(result) { case ((_, res), _) => res }
+  }
+
+  override def resolveFieldsPar(tpe: ObjectType[Ctx, _], value: Any, fields: CollectedFields)(
+      scheme: ExecutionScheme): scheme.Result[Ctx, marshaller.Node] =
+    delegate
+      .resolveFieldsPar(tpe, value, fields)(scheme)
+      .asInstanceOf[scheme.Result[Ctx, marshaller.Node]]
+
+  override def resolveFieldsSeq(tpe: ObjectType[Ctx, _], value: Any, fields: CollectedFields)(
+      scheme: ExecutionScheme): scheme.Result[Ctx, marshaller.Node] =
+    delegate
+      .resolveFieldsSeq(tpe, value, fields)(scheme)
+      .asInstanceOf[scheme.Result[Ctx, marshaller.Node]]
+
+  override def resolveFieldsSubs(tpe: ObjectType[Ctx, _], value: Any, fields: CollectedFields)(
+      scheme: ExecutionScheme): scheme.Result[Ctx, marshaller.Node] =
+    delegate
+      .resolveFieldsSubs(tpe, value, fields)(scheme)
+      .asInstanceOf[scheme.Result[Ctx, marshaller.Node]]
+}

modules/test-cats-effect/src/test/scala/sangria/execution/IOExecutionScheme.scala → modules/cats-effect-experimental/src/test/scala/sangria/catseffect/IOExecutionSchemeSpec.scala

@@ -1,32 +1,23 @@
-package sangria.execution
+package sangria.catseffect
 
 import cats.effect.IO
 import cats.effect.unsafe.implicits.global
 import io.circe.Json
 import org.scalatest.matchers.must.Matchers
 import org.scalatest.wordspec.AnyWordSpec
+import sangria.execution.Executor
+import sangria.catseffect.execution.IOExecutionScheme._
 import sangria.macros._
 import sangria.marshalling.circe._
 import sangria.schema._
 
-import scala.concurrent.{ExecutionContext, Future}
-
 /** The integration with [[cats.effect.IO]] is far from being complete for now.
   */
-class IOExecutionScheme extends AnyWordSpec with Matchers {
-  private implicit val ec: ExecutionContext = global.compute
-  private val ioEffectOps = new EffectOps[IO] {
-    override def failed[Ctx, Res](error: Throwable): IO[Res] = IO.raiseError(error)
-    override def flatMapFuture[Res, T](future: Future[T])(resultFn: T => IO[Res]): IO[Res] =
-      IO.fromFuture(IO(future)).flatMap(resultFn)
-    override def map[T, Out](in: Future[T])(f: T => Out): IO[Out] = IO.fromFuture(IO(in)).map(f)
-  }
-  private implicit val ioExecutionScheme: EffectBasedExecutionScheme[IO] =
-    new EffectBasedExecutionScheme[IO](ioEffectOps, FutureResolverBuilder)
+class IOExecutionSchemeSpec extends AnyWordSpec with Matchers {
 
-  import IOExecutionScheme._
+  import IOExecutionSchemeSpec._
   "IOExecutionScheme" must {
-    "allow using IO effect" in {
+    "allow using IO effect with pure resolve" in {
       val query = gql"""
         query q1 {
           ids
@@ -44,16 +35,35 @@ class IOExecutionScheme extends AnyWordSpec with Matchers {
       )
       res.unsafeRunSync() must be(expected)
     }
+
+    "allow using IO effect with IO resolve" in {
+      val query =
+        gql"""
+          query q1 {
+            parent
+          }
+        """
+      val res: IO[Json] = Executor.execute(schema, query)
+
+      val expected: Json = Json.obj(
+        "data" -> Json.obj(
+          "parent" -> Json.fromString("hello")
+        )
+      )
+      res.unsafeRunSync() must be(expected)
+    }
   }
 }
 
-object IOExecutionScheme {
+object IOExecutionSchemeSpec {
+  import sangria.catseffect.schema.AsyncValue._
   private val QueryType: ObjectType[Unit, Unit] = ObjectType(
     "Query",
     () =>
       fields[Unit, Unit](
-        Field("ids", ListType(IntType), resolve = _ => List(1, 2))
+        Field("ids", ListType(IntType), resolve = _ => List(1, 2)),
+        Field("parent", StringType, resolve = _ => IO("hello"))
       ))
 
-  val schema = Schema(QueryType)
+  private val schema = Schema(QueryType)
 }