Merge pull request xebia-functional#159 from 47deg/optional-fetches

Introduce Fetch#optional for performing optional fetches

docs/src/main/tut/docs.md

@@ -145,13 +145,22 @@ object UserSource extends DataSource[UserId, User]{
 ```
 
 Now that we have a data source we can write a function for fetching users
-given an id, we just have to pass a `UserId` as an argument to `Fetch`.
+given an id, we just have to pass a `UserId` and the data source as arguments to `Fetch`.
 
 ```tut:silent
 def getUser[F[_] : ConcurrentEffect](id: UserId): Fetch[F, User] =
   Fetch(id, UserSource)
 ```
 
+### Optional identities
+
+If you want to create a Fetch that doesn't fail if the identity is not found, you can use `Fetch#optional` instead of `Fetch#apply`. Note that instead of a `Fetch[F, A]` you will get a `Fetch[F, Option[A]]`.
+
+```tut:silent
+def maybeGetUser[F[_] : ConcurrentEffect](id: UserId): Fetch[F, Option[User]] =
+  Fetch.optional(id, UserSource)
+```
+
 ### Data sources that don't support batching
 
 If your data source doesn't support batching, you can simply leave the `batch` method unimplemented. Note that it will use the `fetch` implementation for requesting identities in parallel.

shared/src/main/scala/fetch.scala

@@ -241,6 +241,25 @@ object `package` {
         ))
       )
 
+    def optional[F[_] : ConcurrentEffect, I, A](id: I, ds: DataSource[I, A]): Fetch[F, Option[A]] =
+      Unfetch[F, Option[A]](
+        for {
+          deferred <- Deferred[F, FetchStatus]
+          request = FetchOne(id, ds)
+          result = deferred.complete _
+          blocked = BlockedRequest(request, result)
+          anyDs = ds.asInstanceOf[DataSource[Any, Any]]
+          blockedRequest = RequestMap(Map(anyDs -> blocked))
+        } yield Blocked(blockedRequest, Unfetch[F, Option[A]](
+          deferred.get.map {
+            case FetchDone(a) =>
+              Done(Some(a)).asInstanceOf[FetchResult[F, Option[A]]]
+            case FetchMissing() =>
+              Done(Option.empty[A])
+          }
+        ))
+      )
+
     // Running a Fetch
 
     /**

shared/src/test/scala/FetchTests.scala

@@ -18,16 +18,15 @@ package fetch
 
 import org.scalatest.{AsyncFreeSpec, Matchers}
 
-import fetch._
-
 import scala.concurrent._
 import scala.concurrent.duration._
 
 import cats._
+import cats.temp.par._
 import cats.effect._
 import cats.instances.list._
+import cats.instances.option._
 import cats.data.NonEmptyList
-import cats.syntax.cartesian._
 import cats.syntax.all._
 
 class FetchTests extends AsyncFreeSpec with Matchers {
@@ -717,4 +716,65 @@ class FetchTests extends AsyncFreeSpec with Matchers {
         case Left(MissingIdentity(Never(), _, _)) =>
       }).unsafeToFuture
   }
+
+  // Optional fetches
+
+  case class MaybeMissing(id: Int)
+
+  object MaybeMissingSource extends DataSource[MaybeMissing, Int] {
+    override def name = "Maybe Missing Source"
+
+    override def fetch[F[_]](id: MaybeMissing)(
+      implicit CF: ConcurrentEffect[F], P: Par[F]
+    ): F[Option[Int]] =
+      if (id.id % 2 == 0)
+        Applicative[F].pure(None)
+      else
+        Applicative[F].pure(Option(id.id))
+  }
+
+  def maybeOpt[F[_] : ConcurrentEffect](id: Int): Fetch[F, Option[Int]] =
+    Fetch.optional(MaybeMissing(id), MaybeMissingSource)
+
+  "We can run optional fetches" in {
+    def fetch[F[_] : ConcurrentEffect]: Fetch[F, Option[Int]] =
+      maybeOpt(1)
+
+    Fetch.run[IO](fetch).map(_ shouldEqual Some(1)).unsafeToFuture
+  }
+
+  "We can run optional fetches with traverse" in {
+    def fetch[F[_] : ConcurrentEffect]: Fetch[F, List[Int]] =
+      List(1, 2, 3).traverse(maybeOpt[F]).map(_.flatten)
+
+    Fetch.run[IO](fetch).map(_ shouldEqual List(1, 3)).unsafeToFuture
+  }
+
+  "We can run optional fetches with other data sources" in {
+    def fetch[F[_] : ConcurrentEffect]: Fetch[F, List[Int]] = {
+      val ones = List(1, 2, 3).traverse(one[F])
+      val maybes = List(1, 2, 3).traverse(maybeOpt[F])
+      (ones, maybes).mapN { case (os, ms) => os ++ ms.flatten }
+    }
+
+    Fetch.run[IO](fetch).map(_ shouldEqual List(1, 2, 3, 1, 3)).unsafeToFuture
+  }
+
+  "We can make fetches that depend on optional fetch results when they aren't defined" in {
+    def fetch[F[_] : ConcurrentEffect]: Fetch[F, Int] = for {
+      maybe <- maybeOpt(2)
+      result <- maybe.fold(Fetch.pure(42))(i => one(i))
+    } yield result
+
+    Fetch.run[IO](fetch).map(_ shouldEqual 42).unsafeToFuture
+  }
+
+  "We can make fetches that depend on optional fetch results when they are defined" in {
+    def fetch[F[_] : ConcurrentEffect]: Fetch[F, Int] = for {
+      maybe <- maybeOpt(1)
+      result <- maybe.fold(Fetch.pure(42))(i => one(i))
+    } yield result
+
+    Fetch.run[IO](fetch).map(_ shouldEqual 1).unsafeToFuture
+  }
 }