Merge pull request #612 from sangria-graphql/config-fetcher-concurrency

support for configuring max concurrent batches in fetcher

.scalafmt.conf

@@ -9,6 +9,12 @@ align.preset = none
 
 danglingParentheses.preset = false
 
+rewriteTokens {
+  "⇒":"=>"
+  "←":"<-"
+  "→":"->"
+}
+
 rewrite.rules = [
   AvoidInfix
   RedundantBraces

build.sbt

@@ -21,7 +21,8 @@ ThisBuild / githubWorkflowBuildPreamble ++= List(
 // Binary Incompatible Changes, we'll document.
 ThisBuild / mimaBinaryIssueFilters ++= Seq(
   ProblemFilters.exclude[Problem]("sangria.schema.ProjectedName*"),
-  ProblemFilters.exclude[Problem]("sangria.schema.Args*")
+  ProblemFilters.exclude[Problem]("sangria.schema.Args*"),
+  ProblemFilters.exclude[Problem]("sangria.execution.deferred.FetcherConfig*")
 )
 
 lazy val root = project

modules/core/src/main/scala/sangria/execution/deferred/Fetcher.scala

@@ -201,11 +201,13 @@ object Fetcher {
 
 case class FetcherConfig(
     cacheConfig: Option[() => FetcherCache] = None,
-    maxBatchSizeConfig: Option[Int] = None) {
+    maxBatchSizeConfig: Option[Int] = None,
+    maxConcurrentBatchesConfig: Option[Int] = None) {
   def caching = copy(cacheConfig = Some(() => FetcherCache.simple))
   def caching(cache: FetcherCache) = copy(cacheConfig = Some(() => cache))
 
   def maxBatchSize(size: Int) = copy(maxBatchSizeConfig = Some(size))
+  def maxConcurrentBatches(numBatches: Int) = copy(maxConcurrentBatchesConfig = Some(numBatches))
 }
 
 object FetcherConfig {
@@ -215,6 +217,7 @@ object FetcherConfig {
   def caching(cache: FetcherCache) = empty.caching(cache)
 
   def maxBatchSize(size: Int) = empty.maxBatchSize(size)
+  def maxConcurrentBatches(numBatches: Int) = empty.maxConcurrentBatches(numBatches)
 }
 
 trait DeferredOne[+T, Id] extends Deferred[T] {

modules/core/src/main/scala/sangria/execution/deferred/FetcherBasedDeferredResolver.scala

@@ -182,6 +182,42 @@ class FetcherBasedDeferredResolver[-Ctx](
     }
   }
 
+  // This is here for scala 2.12 / 2.13 compat, as `Iterator.empty` does NOT
+  // take type parameters in 2.12, but does in 2.13.
+  private val emptyIterator = Iterator[Future[(Vector[Any], Try[Seq[Any]])]]()
+
+  private def resolveConcurrentBatches(
+      ctx: FetcherContext[Ctx] @uncheckedVariance,
+      f: Fetcher[Ctx, Any, Any, Any] @uncheckedVariance,
+      nonCachedIds: Vector[Any]
+  )(implicit ec: ExecutionContext): Iterator[Future[(Vector[Any], Try[Seq[Any]])]] = {
+
+    val groupedIds = ctx.fetcher.config.maxBatchSizeConfig match {
+      case Some(size) => nonCachedIds.grouped(size)
+      case None => Iterator(nonCachedIds)
+    }
+
+    val groupedBatches =
+      ctx.fetcher.config.maxConcurrentBatchesConfig match {
+        case Some(size) => groupedIds.grouped(size)
+        case None => Iterator(groupedIds)
+      }
+
+    groupedBatches
+      .foldLeft(emptyIterator) { (accumFutureSeq, groupedIds) =>
+        val results: Iterator[Future[(Vector[Any], Try[Seq[Any]])]] =
+          groupedIds.toIterator.collect {
+            case group if group.nonEmpty =>
+              f.fetch(ctx, group)
+                .map(r => group -> Success(r): (Vector[Any], Try[Seq[Any]]))
+                .recover { case e => group -> Failure(e) }
+            case group =>
+              Future.successful(group -> Success(Seq.empty))
+          }
+        accumFutureSeq ++ results
+      }
+  }
+
   private def resolveEntities(
       ctx: FetcherContext[Ctx] @uncheckedVariance,
       deferredToResolve: Vector[Deferred[Any]],
@@ -191,19 +227,7 @@ class FetcherBasedDeferredResolver[-Ctx](
     val ids = ctx.fetcher.ids(deferredToResolve)
     val (nonCachedIds, cachedResults) = partitionCached(ctx.cache, ids)
 
-    val groupedIds = ctx.fetcher.config.maxBatchSizeConfig match {
-      case Some(size) => nonCachedIds.grouped(size)
-      case None => Iterator(nonCachedIds)
-    }
-
-    val results = groupedIds.map { group =>
-      if (group.nonEmpty)
-        f.fetch(ctx, group).map(r => group -> Success(r): (Vector[Any], Try[Seq[Any]])).recover {
-          case e => group -> Failure(e)
-        }
-      else
-        Future.successful(group -> Success(Seq.empty))
-    }
+    val results = resolveConcurrentBatches(ctx, f, nonCachedIds)
 
     val futureRes = Future.sequence(results).map { allResults =>
       val byId =

modules/core/src/main/scala/sangria/schema/Context.scala

@@ -570,7 +570,8 @@ object Args {
 
     apply(
       schemaElem.arguments,
-      ast.ObjectValue(astElem.arguments.map(arg ⇒ ast.ObjectField(arg.name, arg.value))): ast.Value,
+      ast.ObjectValue(
+        astElem.arguments.map(arg => ast.ObjectField(arg.name, arg.value))): ast.Value,
       Some(variables)
     )
   }