Add Monix task utilities

Author: sebaciv

core/src/main/scala/com/avsystem/commons/concurrent/ObservableExtensions.scala

@@ -22,6 +22,11 @@ object ObservableExtensions extends ObservableExtensions {
       */
     def headOptL: Task[Opt[T]] = obs.headOptionL.map(_.toOpt)
 
+    /**
+      * Returns a [[monix.eval.Task Task]] which emits the first item for which the predicate holds.
+      */
+    def findOptL(p: T => Boolean): Task[Opt[T]] = obs.findL(p).map(_.toOpt)
+
     /** Suppress the duplicate elements emitted by the source Observable.
       *
       * WARNING: this requires unbounded buffering.
@@ -79,5 +84,15 @@ object ObservableExtensions extends ObservableExtensions {
       obs
         .foldLeftL(factory.newBuilder)(_ += _)
         .map(_.result())
+
+    /** Returns a [[monix.eval.Task Task]] that upon evaluation
+      * will collect all items from the source into a [[Map]] instance
+      * using provided functions to compute keys and values.
+      *
+      * WARNING: for infinite streams the process will eventually blow up
+      * with an out of memory error.
+      */
+    def mkMapL[K, V](keyFun: T => K, valueFun: T => V): Task[Map[K, V]] =
+      obs.foldLeftL(Map.newBuilder[K, V])({ case (res, a) => res += ((keyFun(a), valueFun(a))) }).map(_.result())
   }
 }

core/src/main/scala/com/avsystem/commons/concurrent/TaskExtensions.scala

@@ -0,0 +1,57 @@
+package com.avsystem.commons
+package concurrent
+
+import com.avsystem.commons.concurrent.TaskExtensions.{TaskCompanionOps, TaskOps}
+import com.avsystem.commons.misc.Timestamp
+import monix.eval.Task
+
+import java.util.concurrent.TimeUnit
+import scala.concurrent.TimeoutException
+import scala.concurrent.duration.FiniteDuration
+
+trait TaskExtensions {
+  implicit def taskOps[T](task: Task[T]): TaskOps[T] = new TaskOps(task)
+
+  implicit def taskCompanionOps(task: Task.type): TaskCompanionOps.type = TaskCompanionOps
+}
+
+object TaskExtensions extends TaskExtensions {
+  final class TaskOps[T](private val task: Task[T]) extends AnyVal {
+    /**
+      * Like regular `timeout` but [[TimeoutException]] is created lazily (for performance).
+      */
+    def lazyTimeout(after: FiniteDuration, msg: => String): Task[T] =
+      task.timeoutTo(after, Task.raiseError(new TimeoutException(msg)))
+
+    /**
+      * Similar to [[Task.tapEval]], accepts simple consumer function as an argument
+      */
+    def tapL(f: T => Unit): Task[T] =
+      task.map(_.setup(f))
+
+    /**
+      * Similar to [[Task.tapError]], accepts [[PartialFunction]] as an argument
+      */
+    def tapErrorL[B](f: PartialFunction[Throwable, B]): Task[T] =
+      task.tapError(t => Task(f.applyOpt(t)))
+  }
+
+  object TaskCompanionOps {
+    /** A [[Task]] of [[Opt.Empty]] */
+    def optEmpty[A]: Task[Opt[A]] = Task.pure(Opt.Empty)
+
+    def traverseOpt[A, B](opt: Opt[A])(f: A => Task[B]): Task[Opt[B]] =
+      opt.fold(Task.optEmpty[B])(a => f(a).map(_.opt))
+
+    def fromOpt[A](maybeTask: Opt[Task[A]]): Task[Opt[A]] = maybeTask match {
+      case Opt(task) => task.map(_.opt)
+      case Opt.Empty => optEmpty
+    }
+
+    def currentTimestamp: Task[Timestamp] =
+      Task.clock.realTime(TimeUnit.MILLISECONDS).map(Timestamp(_))
+
+    def usingNow[T](useNow: Timestamp => Task[T]): Task[T] =
+      currentTimestamp.flatMap(useNow)
+  }
+}

core/src/test/scala/com/avsystem/commons/concurrent/ObservableExtensionsTest.scala

@@ -20,11 +20,19 @@ class ObservableExtensionsTest extends AnyFunSuite with Matchers
       Observable.fromIterable(ints).headOptL.runToFuture.futureValue shouldBe ints.headOpt
     }
   }
+
+  test("findOptL") {
+    forAll { ints: List[Int] =>
+      Observable.fromIterable(ints).findOptL(_ > 1).runToFuture.futureValue shouldBe ints.findOpt(_ > 1)
+    }
+  }
+
   test("distinct") {
     forAll { ints: List[Int] =>
       Observable.fromIterable(ints).distinct.toListL.runToFuture.futureValue shouldBe ints.distinct
     }
   }
+
   test("distinctBy") {
     forAll { ints: List[Int] =>
       val f: Int => Int = _ % 256
@@ -33,17 +41,20 @@ class ObservableExtensionsTest extends AnyFunSuite with Matchers
         ints.foldLeft(MLinkedHashMap.empty[Int, Int])((map, v) => f(v) |> (key => map.applyIf(!_.contains(key))(_ += key -> v))).valuesIterator.toList
     }
   }
+
   test("sortedL") {
     forAll { ints: List[Int] =>
       Observable.fromIterable(ints).sortedL.runToFuture.futureValue shouldBe ints.sorted
     }
   }
+
   test("sortedByL") {
     forAll { ints: List[Int] =>
       val f: Int => Int = _ % 256
       Observable.fromIterable(ints).sortedByL(f).runToFuture.futureValue shouldBe ints.sortBy(f)
     }
   }
+
   test("toL") {
     forAll { ints: List[(Int, Int)] =>
       def testFactory[T](factory: Factory[(Int, Int), T])(implicit position: Position) =
@@ -78,4 +89,9 @@ class ObservableExtensionsTest extends AnyFunSuite with Matchers
     }
   }
 
+  test("mkMapL") {
+    forAll { ints: List[Int] =>
+      Observable.fromIterable(ints).mkMapL(_ % 3, _ + 2).runToFuture.futureValue shouldBe ints.mkMap(_ % 3, _ + 2)
+    }
+  }
 }