From f173dfdf60d8a758e0db675b1a463ca1007c7c35 Mon Sep 17 00:00:00 2001 From: peterneyens Date: Thu, 3 Nov 2016 12:41:39 +0100 Subject: [PATCH] Refactor and use some more cats features - Refactor interpreter: split out handling of `FetchOne`, `FetchMany` and `Concurrent`, use `XorT` in `processMany` to reduce nested folds, use `Validated` in `processMany` and `processConcurrent` to get missing ids/identities or some result. - (Prematurely ?) optimize `Fetch.join` - Replaced custom implementations of `map2`, `sequence` and `traverse` by using `Applicative[Fetch]` - ... I moved some methods around in the `Fetch` object and with the extraction of the three methods in the interpreter, so the diff looks much bigger than it actually is. --- shared/src/main/scala/fetch.scala | 223 ++++++++-------- shared/src/main/scala/interpreters.scala | 326 ++++++++++++----------- shared/src/test/scala/FetchTests.scala | 2 +- 3 files changed, 277 insertions(+), 274 deletions(-) diff --git a/shared/src/main/scala/fetch.scala b/shared/src/main/scala/fetch.scala index ada1d4a3..e176ad26 100644 --- a/shared/src/main/scala/fetch.scala +++ b/shared/src/main/scala/fetch.scala @@ -22,7 +22,10 @@ import cats.{Applicative, Monad, ApplicativeError, MonadError, ~>, Eval, Recursi import cats.data.{StateT, Const, NonEmptyList, Writer, XorT} import cats.free.Free import cats.instances.list._ +import cats.instances.map._ import cats.instances.option._ +import cats.syntax.foldable._ +import cats.syntax.functor._ import cats.syntax.traverse._ import scala.concurrent.duration.Duration @@ -139,6 +142,12 @@ object `package` { def ap[A, B](ff: Fetch[A => B])(fa: Fetch[A]): Fetch[B] = Fetch.join(ff, fa).map({ case (f, a) => f(a) }) + + override def product[A, B](fa: Fetch[A], fb: Fetch[B]): Fetch[(A, B)] = + Fetch.join(fa, fb) + + override def tuple2[A, B](fa: Fetch[A], fb: Fetch[B]): Fetch[(A, B)] = + Fetch.join(fa, fb) } object Fetch extends FetchInterpreters { @@ -165,45 +174,104 @@ object `package` { ): Fetch[A] = Free.liftF(FetchOne[I, A](i, DS)) - private[this] def combineDeps(ds: List[FetchQuery[_, _]]): List[FetchQuery[_, _]] = { - ds.foldLeft(Map.empty[DataSource[_, _], NonEmptyList[Any]])((acc, op) => - op match { - case one @ FetchOne(id, ds) => - acc.updated(ds, - acc - .get(ds) - .fold(NonEmptyList(id, Nil): NonEmptyList[Any])(accids => { - val newIds = List(id) - val allIds = (accids.toList ++ newIds).distinct - NonEmptyList(allIds.head, allIds.tail) - })) - case many @ FetchMany(ids, ds) => - acc.updated(ds, - acc - .get(ds) - .fold(ids.asInstanceOf[NonEmptyList[Any]])(accids => { - val accList = accids.toList - val newList = ids.toList - val allIds = (accList ++ newList).distinct - NonEmptyList(allIds.head, allIds.tail) - })) - case _ => acc - }) - .toList - .map({ - case (ds, ids) if ids.toList.size == 1 => - FetchOne[Any, Any](ids.head, ds.asInstanceOf[DataSource[Any, Any]]) - case (ds, ids) => - FetchMany[Any, Any](ids, ds.asInstanceOf[DataSource[Any, Any]]) - }) + /** + * Given a list of `FetchRequest`s, lift it to the `Fetch` monad. When executing + * the fetch, data sources will be queried and the fetch will return a `DataSourceCache` + * containing the results. + */ + private[this] def concurrently(fetches: List[FetchQuery[_, _]]): Fetch[DataSourceCache] = + Free.liftF(Concurrent(fetches)) + + /** + * Transform a list of fetches into a fetch of a list. It implies concurrent execution of fetches. + */ + def sequence[I, A](ids: List[Fetch[A]]): Fetch[List[A]] = + Applicative[Fetch].sequence(ids) + + /** + * Apply a fetch-returning function to every element in a list and return a Fetch of the list of + * results. It implies concurrent execution of fetches. + */ + def traverse[A, B](ids: List[A])(f: A => Fetch[B]): Fetch[List[B]] = + Applicative[Fetch].traverse(ids)(f) + + /** + * Apply the given function to the result of the two fetches. It implies concurrent execution of fetches. + */ + def map2[A, B, C](f: (A, B) => C)(fa: Fetch[A], fb: Fetch[B]): Fetch[C] = + Applicative[Fetch].map2(fa, fb)(f) + + /** + * Join two fetches from any data sources and return a Fetch that returns a tuple with the two + * results. It implies concurrent execution of fetches. + */ + def join[A, B](fl: Fetch[A], fr: Fetch[B]): Fetch[(A, B)] = { + def depFetches(fa: Fetch[_], fb: Fetch[_]): List[FetchQuery[_, _]] = + combineQueries(dependentQueries(fa) ++ dependentQueries(fb)) + + def joinWithFetches( + fl: Fetch[A], fr: Fetch[B], fetches: List[FetchQuery[_, _]]): Fetch[(A, B)] = + concurrently(fetches).flatMap(cache => joinH(fl, fr, cache)) + + def joinH(fl: Fetch[A], fr: Fetch[B], cache: DataSourceCache): Fetch[(A, B)] = { + val sfl = fl.compile(simplify(cache)) + val sfr = fr.compile(simplify(cache)) + + val remainingDeps = depFetches(sfl, sfr) + + if (remainingDeps.isEmpty) Monad[Fetch].tuple2(sfl, sfr) + else joinWithFetches(sfl, sfr, remainingDeps) + } + + joinWithFetches(fl, fr, depFetches(fl, fr)) } - private[this] type FM = List[FetchOp[_]] - private[this] type KeepFetches[A] = Writer[FM, A] + /** + * Use a `DataSourceCache` to optimize a `FetchOp`. + * If the cache contains all the fetch identities, the fetch doesn't need to be + * executed and can be replaced by cached results. + */ + private[this] def simplify(cache: DataSourceCache): (FetchOp ~> FetchOp) = { + new (FetchOp ~> FetchOp) { + def apply[B](fetchOp: FetchOp[B]): FetchOp[B] = fetchOp match { + case one @ FetchOne(id, ds) => + cache.get[B](ds.identity(id)).fold(fetchOp)(b => Fetched(b)) + case many @ FetchMany(ids, ds) => + val fetched = ids.traverse(id => cache.get(ds.identity(id))) + fetched.fold(fetchOp)(results => Fetched(results.toList)) + case conc @ Concurrent(manies) => + val newManies = manies.filterNot(_.fullfilledBy(cache)) + (if (newManies.isEmpty) Fetched(cache) else Concurrent(newManies)): FetchOp[B] + case other => other + } + } + } + + /** + * Combine multiple queries so the resulting `List` only contains one `FetchQuery` + * per `DataSource`. + */ + private[this] def combineQueries(ds: List[FetchQuery[_, _]]): List[FetchQuery[_, _]] = + ds.foldMap[Map[DataSource[_, _], NonEmptyList[Any]]] { + case FetchOne(id, ds) => Map(ds -> NonEmptyList.of[Any](id)) + case FetchMany(ids, ds) => Map(ds -> ids.widen[Any]) + } + .mapValues { nel => + // workaround because NEL[Any].distinct needs Order[Any] + NonEmptyList.fromListUnsafe(nel.toList.distinct) + } + .toList + .map { + case (ds, NonEmptyList(id, Nil)) => FetchOne(id, ds.castDS[Any, Any]) + case (ds, ids) => FetchMany(ids, ds.castDS[Any, Any]) + } + + private[this] type FetchOps = List[FetchOp[_]] + private[this] type KeepFetches[A] = Writer[FetchOps, A] private[this] type AnalyzeTop[A] = XorT[KeepFetches, Unit, A] private[this] object AnalyzeTop { - def stopWith[R](list: FM): AnalyzeTop[R] = + def stopWith[R](list: FetchOps): AnalyzeTop[R] = AnalyzeTop.stop(Writer.tell(list)) def stopEmpty[R]: AnalyzeTop[R] = @@ -216,12 +284,15 @@ object `package` { XorT.right[KeepFetches, Unit, X](k) } - private[this] def deps(f: Fetch[_]): List[FetchQuery[_, _]] = { + /** + * Get a list of dependent `FetchQuery`s for a given `Fetch`. + */ + private[this] def dependentQueries(f: Fetch[_]): List[FetchQuery[_, _]] = { val analyzeTop: FetchOp ~> AnalyzeTop = new (FetchOp ~> AnalyzeTop) { def apply[A](op: FetchOp[A]): AnalyzeTop[A] = op match { case fetc @ Fetched(c) => AnalyzeTop.go(Writer(List(fetc), c)) case one @ FetchOne(_, _) => AnalyzeTop.stopWith(List(one)) - case conc @ Concurrent(as) => AnalyzeTop.stopWith(as.asInstanceOf[FM]) + case conc @ Concurrent(as) => AnalyzeTop.stopWith(as.asInstanceOf[FetchOps]) case _ => AnalyzeTop.stopEmpty } } @@ -232,82 +303,6 @@ object `package` { } } - private[this] def concurrently(fa: Fetch[_], fb: Fetch[_]): Fetch[DataSourceCache] = { - val fetches: List[FetchQuery[_, _]] = combineDeps(deps(fa) ++ deps(fb)) - Free.liftF(Concurrent(fetches)) - } - - /** - * Transform a list of fetches into a fetch of a list. It implies concurrent execution of fetches. - */ - def sequence[I, A](ids: List[Fetch[A]]): Fetch[List[A]] = { - ids.foldLeft(Fetch.pure(List(): List[A]))((f, newF) => - Fetch.join(f, newF).map(t => t._1 :+ t._2)) - } - - /** - * Apply a fetch-returning function to every element in a list and return a Fetch of the list of - * results. It implies concurrent execution of fetches. - */ - def traverse[A, B](ids: List[A])(f: A => Fetch[B]): Fetch[List[B]] = - sequence(ids.map(f)) - - /** - * Apply the given function to the result of the two fetches. It implies concurrent execution of fetches. - */ - def map2[A, B, C](f: (A, B) => C)(fa: Fetch[A], fb: Fetch[B]): Fetch[C] = - Fetch.join(fa, fb).map({ case (a, b) => f(a, b) }) - - private[this] def simplify(results: DataSourceCache): (FetchOp ~> FetchOp) = { - new (FetchOp ~> FetchOp) { - def apply[B](f: FetchOp[B]): FetchOp[B] = f match { - case one @ FetchOne(id, ds) => { - results.get[B](ds.identity(id)).fold(one: FetchOp[B])(b => Fetched(b)) - } - case many @ FetchMany(ids, ds) => { - val fetched = ids.map(id => results.get(ds.identity(id))).toList.sequence - fetched.fold({ - many: FetchOp[B] - })(results => Fetched(results)) - } - case conc @ Concurrent(manies) => { - val newManies = manies.filterNot(_.fullfilledBy(results)) - - if (newManies.isEmpty) - Fetched(results).asInstanceOf[FetchOp[B]] - else - Concurrent(newManies).asInstanceOf[FetchOp[B]] - } - case other => other - } - } - } - - /** - * Join two fetches from any data sources and return a Fetch that returns a tuple with the two - * results. It implies concurrent execution of fetches. - */ - def join[A, B](fl: Fetch[A], fr: Fetch[B]): Fetch[(A, B)] = { - for { - cache <- concurrently(fl, fr) - result <- { - val sfl = fl.compile(simplify(cache)) - val sfr = fr.compile(simplify(cache)) - - val remainingDeps = combineDeps(deps(sfl) ++ deps(sfr)) - - if (remainingDeps.isEmpty) { - for { - a <- sfl - b <- sfr - } yield (a, b) - } else { - join[A, B](sfl, sfr) - } - } - } yield result - } - class FetchRunner[M[_]] { def apply[A]( fa: Fetch[A], @@ -357,4 +352,8 @@ object `package` { */ def run[M[_]]: FetchRunnerA[M] = new FetchRunnerA[M] } + + private[fetch] implicit class DataSourceCast[A, B](ds: DataSource[A, B]) { + def castDS[C, D]: DataSource[C, D] = ds.asInstanceOf[DataSource[C, D]] + } } diff --git a/shared/src/main/scala/interpreters.scala b/shared/src/main/scala/interpreters.scala index ca043f0b..273d2fa2 100644 --- a/shared/src/main/scala/interpreters.scala +++ b/shared/src/main/scala/interpreters.scala @@ -19,182 +19,186 @@ package fetch import scala.collection.immutable._ import cats.{MonadError, ~>} -import cats.data.{StateT, NonEmptyList} +import cats.data.{OptionT, NonEmptyList, StateT, Validated, XorT} import cats.instances.option._ import cats.instances.list._ +import cats.instances.map._ +import cats.syntax.either._ +import cats.syntax.flatMap._ +import cats.syntax.foldable._ +import cats.syntax.functor._ +import cats.syntax.functorFilter._ +import cats.syntax.option._ import cats.syntax.traverse._ +import cats.syntax.validated._ trait FetchInterpreters { def pendingQueries( - queries: List[FetchQuery[_, _]], cache: DataSourceCache): List[FetchQuery[Any, Any]] = { - - queries - .filterNot(_.fullfilledBy(cache)) - .map(req => { - (req.dataSource, req.missingIdentities(cache)) - }) - .collect({ - case (ds, ids) if ids.size == 1 => - FetchOne[Any, Any](ids.head, ds.asInstanceOf[DataSource[Any, Any]]) - case (ds, ids) if ids.size > 1 => - FetchMany[Any, Any]( - NonEmptyList(ids(0), ids.tail), ds.asInstanceOf[DataSource[Any, Any]]) - }) - } + queries: List[FetchQuery[_, _]], + cache: DataSourceCache + ): List[FetchQuery[Any, Any]] = + queries.mapFilter { query => + val dsAny = query.dataSource.castDS[Any, Any] + NonEmptyList.fromList(query.missingIdentities(cache)).map { + case NonEmptyList(id, Nil) => FetchOne(id, dsAny) + case ids => FetchMany(ids.widen[Any], dsAny) + } + } def interpreter[I, M[_]]( implicit M: FetchMonadError[M] ): FetchOp ~> FetchInterpreter[M]#f = { new (FetchOp ~> FetchInterpreter[M]#f) { - def apply[A](fa: FetchOp[A]): FetchInterpreter[M]#f[A] = { + def apply[A](fa: FetchOp[A]): FetchInterpreter[M]#f[A] = StateT[M, FetchEnv, A] { env: FetchEnv => fa match { - case Thrown(e) => M.raiseError(UnhandledException(e)) - case Fetched(a) => M.pure((env, a)) - case one @ FetchOne(id, ds) => { - val startRound = System.nanoTime() - val cache = env.cache - - cache - .get[A](ds.identity(id)) - .fold[M[(FetchEnv, A)]]( - M.flatMap(M.runQuery(ds.fetchOne(id)))((res: Option[A]) => { - val endRound = System.nanoTime() - res.fold[M[(FetchEnv, A)]]( - M.raiseError( - NotFound(env, one) - ) - )(result => { - val endRound = System.nanoTime() - val newCache = cache.update(ds.identity(id), result) - val round = Round(cache, one, result, startRound, endRound) - M.pure(env.evolve(round, newCache) -> result) - }) - }) - )(cached => { - val endRound = System.nanoTime() - M.pure(env -> cached) - }) - } - case many @ FetchMany(ids, ds) => { - val startRound = System.nanoTime() - val cache = env.cache - val newIds = many.missingIdentities(cache) - val result = ids.toList.flatMap(id => cache.get(ds.identity(id))) - - if (newIds.isEmpty) - M.pure(env -> result) - else { - M.flatMap(M.runQuery(ds - .asInstanceOf[DataSource[I, A]] - .fetchMany(NonEmptyList(newIds(0).asInstanceOf[I], - newIds.tail.asInstanceOf[List[I]]))))( - (res: Map[I, A]) => { - val endRound = System.nanoTime() - - ids.toList - .map(i => res.get(i.asInstanceOf[I])) - .sequence - .fold[M[(FetchEnv, A)]]({ - val missingIdentities = ids.toList - .map(i => i.asInstanceOf[I] -> res.get(i.asInstanceOf[I])) - .collect({ - case (i, None) => i - }) - M.raiseError( - MissingIdentities(env, Map(ds.name -> missingIdentities)) - ) - })(results => { - val endRound = System.nanoTime() - val newCache = - cache.cacheResults[I, A](res, ds.asInstanceOf[DataSource[I, A]]) - val round = Round(cache, many, results, startRound, endRound) - M.pure(env.evolve(round, newCache) -> results.asInstanceOf[A]) - }) - }) - } - } - - case conc @ Concurrent(concurrentQueries) => { - val startRound = System.nanoTime() - val cache = env.cache - - val queries: List[FetchQuery[Any, Any]] = pendingQueries(concurrentQueries, cache) - - if (queries.isEmpty) - M.pure((env, cache.asInstanceOf[A])) - else { - val sentQueries = M.sequence(queries.map({ - case FetchOne(a, ds) => { - val ident = a.asInstanceOf[I] - val task = M.runQuery(ds.asInstanceOf[DataSource[I, A]].fetchOne(ident)) - M.map(task)((r: Option[A]) => - r.fold(Map.empty[I, A])((result: A) => Map(ident -> result))) - } - case FetchMany(as, ds) => - M.runQuery(ds - .asInstanceOf[DataSource[I, A]] - .fetchMany(as.asInstanceOf[NonEmptyList[I]])) - })) - - M.flatMap(sentQueries)((results: List[Map[_, _]]) => { - val endRound = System.nanoTime() - val newCache = (queries zip results).foldLeft(cache)((accache, resultset) => { - val (req, resultmap) = resultset - val ds = req.dataSource - val tresults = resultmap.asInstanceOf[Map[I, A]] - val tds = ds.asInstanceOf[DataSource[I, A]] - accache.cacheResults[I, A](tresults, tds) - }) - - val allFullfilled = (queries zip results).forall({ - case (FetchOne(_, _), results) => results.size == 1 - case (FetchMany(as, _), results) => as.toList.size == results.size - case _ => false - }) - - if (allFullfilled) { - val round = Round( - cache, - Concurrent(queries), - results, - startRound, - endRound - ) - val newEnv = env.evolve(round, newCache) - // since user-provided caches may discard elements, we use an in-memory - // cache to gather these intermediate results that will be used for - // concurrent optimizations. - val cachedResults = - (queries zip results).foldLeft(InMemoryCache.empty)((cach, resultSet) => { - val (req, resultmap) = resultSet - val ds = req.dataSource - val tresults = resultmap.asInstanceOf[Map[I, A]] - val tds = ds.asInstanceOf[DataSource[I, A]] - cach.cacheResults[I, A](tresults, tds).asInstanceOf[InMemoryCache] - }) - - M.pure((newEnv, cachedResults.asInstanceOf[A])) - } else { - val missingIdentities: Map[DataSourceName, List[Any]] = (queries zip results) - .collect({ - case (FetchOne(id, ds), results) if results.size != 1 => - ds.name -> List(id) - case (FetchMany(as, ds), results) if results.size != as.toList.size => - ds.name -> as.toList.collect({ - case i if !results.asInstanceOf[Map[Any, Any]].get(i).isDefined => i - }) - }) - .toMap - M.raiseError( - MissingIdentities(env, missingIdentities) - ) - } - }) - } - } + case Thrown(e) => M.raiseError(UnhandledException(e)) + case Fetched(a) => M.pure((env, a)) + case one @ FetchOne(_, _) => processOne(one, env) + case many @ FetchMany(_, _) => processMany(many, env) + case conc @ Concurrent(_) => processConcurrent(conc, env) } } + } + } + + private[this] def processOne[M[_], A]( + one: FetchOne[Any, A], + env: FetchEnv + )( + implicit M: FetchMonadError[M] + ): M[(FetchEnv, A)] = { + val FetchOne(id, ds) = one + val startRound = System.nanoTime() + env.cache + .get[A](ds.identity(id)) + .fold[M[(FetchEnv, A)]]( + M.runQuery(ds.fetchOne(id)).flatMap { (res: Option[A]) => + val endRound = System.nanoTime() + res.fold[M[(FetchEnv, A)]] { + // could not get result from datasource + M.raiseError(NotFound(env, one)) + } { result => + // found result (and update cache) + val newCache = env.cache.update(ds.identity(id), result) + val round = Round(env.cache, one, result, startRound, endRound) + M.pure(env.evolve(round, newCache) -> result) + } + } + ) { cached => + // get result from cache + M.pure(env -> cached) + } + } + + private[this] def processMany[M[_], A]( + many: FetchMany[Any, Any], + env: FetchEnv + )( + implicit M: FetchMonadError[M], + ev: List[Any] =:= A + ): M[(FetchEnv, A)] = { + val ids = many.as + val ds = many.ds //.castDS[Any, Any] + val startRound = System.nanoTime() + val cache = env.cache + val newIds = many.missingIdentities(cache) + + (for { + newIdsNel <- XorT.fromXor[M] { + NonEmptyList.fromList(newIds).toRightXor { + // no missing ids, get all from cache + val cachedResults = ids.toList.mapFilter(id => cache.get(ds.identity(id))) + env -> cachedResults + } + } + resMap <- XorT.right(M.runQuery(ds.fetchMany(newIdsNel))) + results <- ids.toList + .traverseU(id => resMap.get(id).toValidNel(id)) + .fold[XorT[M, (FetchEnv, List[Any]), List[Any]]]({ missingIds => + // not all identities could be found + val map = Map(ds.name -> missingIds.toList) + XorT.left(M.raiseError(MissingIdentities(env, map))) + }, XorT.pure) + } yield { + // found all results (and update cache) + val endRound = System.nanoTime() + val newCache = cache.cacheResults(resMap, ds) + val round = Round(cache, many, results, startRound, endRound) + env.evolve(round, newCache) -> results + }).merge.map { case (env, l) => (env, ev(l)) } // A =:= List[Any] + } + + private[this] def processConcurrent[M[_]]( + concurrent: Concurrent, + env: FetchEnv + )( + implicit M: FetchMonadError[M] + ): M[(FetchEnv, DataSourceCache)] = { + def runFetchQueryAsMap[I, A](op: FetchQuery[I, A]): M[Map[I, A]] = + op match { + case FetchOne(a, ds) => + OptionT(M.runQuery(ds.fetchOne(a))).map(r => Map(a -> r)).getOrElse(Map.empty) + case FetchMany(as, ds) => M.runQuery(ds.fetchMany(as)) + } + + type MissingIdentitiesMap = Map[DataSourceName, List[Any]] + + // Give for a list of queries and result(maps) all the missing identities + def missingIdentitiesOrAllFulfilled( + queriesAndResults: List[(FetchQuery[Any, Any], Map[Any, Any])] + ): Validated[MissingIdentitiesMap, Unit] = + queriesAndResults.traverseU_ { + case (FetchOne(id, ds), resultMap) => + Either.cond(resultMap.size == 1, (), Map(ds.name -> List(id))).toValidated + case (FetchMany(as, ds), resultMap) => + Either + .cond(as.toList.size == resultMap.size, + (), + Map(ds.name -> as.toList.filter(id => resultMap.get(id).isEmpty))) + .toValidated + case _ => + Map.empty[DataSourceName, List[Any]].invalid + } + + val startRound = System.nanoTime() + val cache = env.cache + + val queries: List[FetchQuery[Any, Any]] = pendingQueries(concurrent.as, cache) + + if (queries.isEmpty) + // there are no pending queries + M.pure((env, cache)) + else { + val sentRequests = queries.traverse(r => runFetchQueryAsMap(r)) + + sentRequests.flatMap { results => + val endRound = System.nanoTime() + val queriesAndResults = queries zip results + + val missingOrFulfilled = missingIdentitiesOrAllFulfilled(queriesAndResults) + + missingOrFulfilled.fold({ missingIds => + // not all identiies were found + M.raiseError(MissingIdentities(env, missingIds)) + }, { _ => + // results found for all identities + val round = Round(cache, Concurrent(queries), results, startRound, endRound) + + // since user-provided caches may discard elements, we use an in-memory + // cache to gather these intermediate results that will be used for + // concurrent optimizations. + val (newCache, cachedResults) = + queriesAndResults.foldLeft((cache, InMemoryCache.empty)) { + case ((userCache, internCache), (req, resultMap)) => + val anyMap = resultMap.asInstanceOf[Map[Any, Any]] + val anyDS = req.dataSource.castDS[Any, Any] + (userCache.cacheResults(anyMap, anyDS), + internCache.cacheResults(anyMap, anyDS).asInstanceOf[InMemoryCache]) + } + + M.pure((env.evolve(round, newCache), cachedResults)) + }) } } } diff --git a/shared/src/test/scala/FetchTests.scala b/shared/src/test/scala/FetchTests.scala index e370b161..afe42978 100644 --- a/shared/src/test/scala/FetchTests.scala +++ b/shared/src/test/scala/FetchTests.scala @@ -60,6 +60,7 @@ object TestHelper { override def fetchMany(ids: NonEmptyList[Many]): Query[Map[Many, List[Int]]] = Query.sync(ids.toList.map(m => (m, 0 until m.n toList)).toMap) } + def many(id: Int): Fetch[List[Int]] = Fetch(Many(id)) case class Never() implicit object NeverSource extends DataSource[Never, Int] { @@ -69,7 +70,6 @@ object TestHelper { override def fetchMany(ids: NonEmptyList[Never]): Query[Map[Never, Int]] = Query.sync(Map.empty[Never, Int]) } - def many(id: Int): Fetch[List[Int]] = Fetch(Many(id)) def requestFetches(r: FetchRequest): Int = r match {