New implementation parallel joins

build.sbt

@@ -29,7 +29,6 @@ lazy val buildSettings = Seq(
 lazy val commonSettings = Seq(
   resolvers += Resolver.sonatypeRepo("releases"),
   libraryDependencies ++= Seq(
-    "org.typelevel" %%% "cats-core" % "0.8.1",
     "org.typelevel" %%% "cats-free" % "0.8.1",
     "org.scalatest" %%% "scalatest" % "3.0.0" % "test",
     compilerPlugin(

shared/src/main/scala/fetch.scala

@@ -61,36 +61,29 @@ object Query {
   }
 }
 
-/** Requests in Fetch Free monad.
-  */
-sealed trait FetchRequest extends Product with Serializable {
-  def fullfilledBy(cache: DataSourceCache): Boolean
-}
+trait FetchException                                   extends Throwable with Product with Serializable
+case class NotFound(env: Env, request: FetchOne[_, _]) extends FetchException
+case class MissingIdentities(env: Env, missing: Map[DataSourceName, List[Any]])
+    extends FetchException
+case class UnhandledException(err: Throwable) extends FetchException
+
+/** Requests in Fetch Free monad. */
+sealed trait FetchRequest extends Product with Serializable
 
 sealed trait FetchQuery[I, A] extends FetchRequest {
   def missingIdentities(cache: DataSourceCache): List[I]
   def dataSource: DataSource[I, A]
   def identities: NonEmptyList[I]
 }
 
-trait FetchException                                   extends Throwable with Product with Serializable
-case class NotFound(env: Env, request: FetchOne[_, _]) extends FetchException
-case class MissingIdentities(env: Env, missing: Map[DataSourceName, List[Any]])
-    extends FetchException
-case class UnhandledException(err: Throwable) extends FetchException
-
 /**
   * Primitive operations in the Fetch Free monad.
   */
 sealed abstract class FetchOp[A] extends Product with Serializable
 
-final case class Fetched[A](a: A) extends FetchOp[A]
 final case class FetchOne[I, A](id: I, ds: DataSource[I, A])
     extends FetchOp[A]
     with FetchQuery[I, A] {
-  override def fullfilledBy(cache: DataSourceCache): Boolean = {
-    cache.contains(ds.identity(id))
-  }
   override def missingIdentities(cache: DataSourceCache): List[I] = {
     cache.get[A](ds.identity(id)).fold(List(id))(_ => Nil)
   }
@@ -101,9 +94,6 @@ final case class FetchOne[I, A](id: I, ds: DataSource[I, A])
 final case class FetchMany[I, A](ids: NonEmptyList[I], ds: DataSource[I, A])
     extends FetchOp[List[A]]
     with FetchQuery[I, A] {
-  override def fullfilledBy(cache: DataSourceCache): Boolean = {
-    ids.forall(i => cache.contains(ds.identity(i)))
-  }
 
   override def missingIdentities(cache: DataSourceCache): List[I] = {
     ids.toList.distinct.filterNot(i => cache.contains(ds.identity(i)))
@@ -113,12 +103,10 @@ final case class FetchMany[I, A](ids: NonEmptyList[I], ds: DataSource[I, A])
 }
 final case class Concurrent(queries: NonEmptyList[FetchQuery[Any, Any]])
     extends FetchOp[InMemoryCache]
-    with FetchRequest {
-  override def fullfilledBy(cache: DataSourceCache): Boolean = {
-    queries.forall(_.fullfilledBy(cache))
-  }
-}
-final case class Thrown[A](err: Throwable) extends FetchOp[A]
+    with FetchRequest
+
+final case class Join[A, B](fl: Fetch[A], fr: Fetch[B]) extends FetchOp[(A, B)]
+final case class Thrown[A](err: Throwable)              extends FetchOp[A]
 
 object `package` {
   type DataSourceName     = String
@@ -139,6 +127,9 @@ object `package` {
     override def product[A, B](fa: Fetch[A], fb: Fetch[B]): Fetch[(A, B)] =
       Fetch.join(fa, fb)
 
+    override def map2[A, B, Z](fa: Fetch[A], fb: Fetch[B])(f: (A, B) => Z): Fetch[Z] =
+      Fetch.join(fa, fb).map { case (a, b) => f(a, b) }
+
     override def tuple2[A, B](fa: Fetch[A], fb: Fetch[B]): Fetch[(A, B)] =
       Fetch.join(fa, fb)
   }
@@ -174,7 +165,7 @@ object `package` {
       Free.liftF(FetchMany(NonEmptyList(i, is.toList), DS))
 
     /**
-      * Given a non empty list of `FetchRequest`s, lift it to the `Fetch` monad. When executing
+      * Given a non empty list of `FetchQuery`s, lift it to the `Fetch` monad. When executing
       * the fetch, data sources will be queried and the fetch will return an `InMemoryCache`
       * containing the results.
       */
@@ -205,109 +196,8 @@ object `package` {
       * 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 parallelizableQueries(fa: Fetch[_], fb: Fetch[_]): List[FetchQuery[_, _]] =
-        combineQueries(independentQueries(fa) ++ independentQueries(fb))
-
-      def parallelizableQueriesAny(fa: Fetch[_], fb: Fetch[_]): List[FetchQuery[Any, Any]] =
-        parallelizableQueries(fa, fb).asInstanceOf[List[FetchQuery[Any, Any]]]
-
-      def joinWithQueries(
-          fl: Fetch[A],
-          fr: Fetch[B],
-          queries: List[FetchQuery[Any, Any]]
-      ): Fetch[(A, B)] = {
-        queries.toNel.fold(Monad[Fetch].tuple2(fl, fr)) { queriesNel =>
-          concurrently(queriesNel).flatMap { cache =>
-            val sfl = fl.compile(simplify(cache))
-            val sfr = fr.compile(simplify(cache))
-
-            val deps = parallelizableQueriesAny(sfl, sfr)
-            // joinWithQueries(sfl, sfr, deps diff fetches)
-            joinWithQueries(sfl, sfr, deps)
-          }
-        }
-      }
-
-      joinWithQueries(fl, fr, parallelizableQueriesAny(fl, fr))
-    }
-
-    /**
-      * 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: InMemoryCache): (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.toList.filterNot(_.fullfilledBy(cache))
-            newManies.toNel.fold[FetchOp[B]](Fetched(cache))(Concurrent(_))
-          case other => other
-        }
-      }
-    }
-
-    /**
-      * Combine multiple queries so the resulting `List` only contains one `FetchQuery`
-      * per `DataSource`.
-      */
-    private[this] def combineQueries(qs: List[FetchQuery[_, _]]): List[FetchQuery[_, _]] =
-      qs.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 would need Order[Any]
-          nel.unsafeListOp(_.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]  = EitherT[KeepFetches, Unit, A]
-
-    private[this] object AnalyzeTop {
-      def stopWith[R](list: FetchOps): AnalyzeTop[R] =
-        AnalyzeTop.stop(Writer.tell(list))
-
-      def stopEmpty[R]: AnalyzeTop[R] =
-        AnalyzeTop.stop(Writer.value(()))
-
-      def stop[R](k: KeepFetches[Unit]): AnalyzeTop[R] =
-        EitherT.left[KeepFetches, Unit, R](k)
-
-      def go[X](k: KeepFetches[X]): AnalyzeTop[X] =
-        EitherT.right[KeepFetches, Unit, X](k)
-    }
-
-    /**
-      * Get a list of independent `FetchQuery`s for a given `Fetch`.
-      */
-    private[this] def independentQueries(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(), c))
-          case one @ FetchOne(_, _)  => AnalyzeTop.stopWith(List(one))
-          case conc @ Concurrent(as) => AnalyzeTop.stopWith(as.toList.asInstanceOf[FetchOps])
-          case _                     => AnalyzeTop.stopEmpty
-        }
-      }
-
-      f.foldMap[AnalyzeTop](analyzeTop).value.written.collect {
-        case one @ FetchOne(_, _)   => one
-        case many @ FetchMany(_, _) => many
-      }
-    }
+    def join[A, B](fl: Fetch[A], fr: Fetch[B]): Fetch[(A, B)] =
+      Free.liftF(Join(fl, fr))
 
     class FetchRunner[M[_]] {
       def apply[A](

shared/src/main/scala/freeinspect.scala

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2016 47 Degrees, LLC. <http://www.47deg.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.free
+
+import cats.{Id, ~>}
+import cats.data.Coproduct
+
+import scala.annotation.tailrec
+
+object FreeTopExt {
+
+  @tailrec
+  def inspect[F[_]](free: Free[F, _]): Option[F[_]] = free match {
+    case Free.Pure(p)              => None
+    case Free.Suspend(fa)          => Some(fa)
+    case Free.FlatMapped(free2, _) => inspect(free2)
+  }
+
+  /**
+    * Is the first step a `Pure` ?
+    *
+    * Could be implemented using `resume` as :
+    * {{{
+    * val liftCoyoneda: F ~> Coyoneda[F, ?] = λ[(F ~> Coyoneda[F, ?])](Coyoneda.lift(_))
+    * free.compile[Coyoneda[F, ?]](liftCoyoneda).resume.toOption
+    * }}}
+    */
+  def inspectPure[F[_], A](free: Free[F, A]): Option[A] = free match {
+    case Free.Pure(p)              => Some(p)
+    case Free.Suspend(fa)          => None
+    case Free.FlatMapped(free2, _) => None
+  }
+
+  def modify[F[_], A](free: Free[F, A])(f: F ~> Coproduct[F, Id, ?]): Free[F, A] =
+    free match {
+      case pure @ Free.Pure(_)          => pure
+      case Free.Suspend(fa)             => f(fa).run.fold(Free.liftF(_), Free.Pure(_))
+      case Free.FlatMapped(free2, cont) => Free.FlatMapped(modify(free2)(f), cont).step
+    }
+
+  def print[F[_], A](free: Free[F, A]): String = free match {
+    case Free.Pure(p)              => s"Pure($p)"
+    case Free.Suspend(fa)          => s"Suspend($fa)"
+    case Free.FlatMapped(free2, _) => s"FlatMapped(${print(free2)}, <fb => Free>)"
+  }
+
+}