Implement Task.traverseMap and unit tests for the new extensions

Author: sebaciv

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

@@ -93,6 +93,6 @@ object ObservableExtensions extends ObservableExtensions {
       * 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())
+      obs.map(v => (keyFun(v), valueFun(v))).toL(Map)
   }
 }

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

@@ -45,9 +45,15 @@ object TaskExtensions extends TaskExtensions {
 
     def fromOpt[A](maybeTask: Opt[Task[A]]): Task[Opt[A]] = maybeTask match {
       case Opt(task) => task.map(_.opt)
-      case Opt.Empty => optEmpty
+      case Opt.Empty => Task.optEmpty
     }
 
+    def traverseMap[K, V, A, B](map: Map[K, V])(f: (K, V) => Task[(A, B)]): Task[Map[A, B]] =
+      Task.traverse(map.toSeq)({ case (key, value) => f(key, value) }).map(_.toMap)
+
+    def traverseMapValues[K, A, B](map: Map[K, A])(f: (K, A) => Task[B]): Task[Map[K, B]] =
+      traverseMap(map)({ case (key, value) => f(key, value).map(key -> _) })
+
     def currentTimestamp: Task[Timestamp] =
       Task.clock.realTime(TimeUnit.MILLISECONDS).map(Timestamp(_))
 

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

@@ -0,0 +1,52 @@
+package com.avsystem.commons
+package concurrent
+
+import monix.eval.Task
+import monix.execution.Scheduler
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
+
+import scala.concurrent.TimeoutException
+import scala.concurrent.duration._
+
+class TaskExtensionsTest extends AnyFunSuite with Matchers with ScalaCheckDrivenPropertyChecks with ScalaFutures {
+  import com.avsystem.commons.concurrent.TaskExtensions._
+
+  private implicit val scheduler: Scheduler = Scheduler(RunNowEC)
+
+  test("lazyTimeout") {
+    val result = Task.never.lazyTimeout(100.millis, "Lazy timeout").runToFuture.failed.futureValue
+    result shouldBe a[TimeoutException]
+    result.getMessage shouldBe "Lazy timeout"
+  }
+
+  test("traverseOpt") {
+    Task.traverseOpt(Opt.empty[Int])(i => Task.now(i)).runToFuture.futureValue shouldBe Opt.Empty
+    Task.traverseOpt(Opt.some(123))(i => Task.now(i)).runToFuture.futureValue shouldBe Opt.some(123)
+  }
+
+  test("fromOpt") {
+    Task.fromOpt(Opt.empty[Task[Int]]).runToFuture.futureValue shouldBe Opt.Empty
+    Task.fromOpt(Opt.some(Task.now(123))).runToFuture.futureValue shouldBe Opt.some(123)
+  }
+
+  test("traverseMap") {
+    forAll { data: List[(String, Int)] =>
+      val map = data.toMap
+      val expected = map.view.map({ case (key, value) => (key + key, value + 2) }).toMap
+      val result = Task.traverseMap(map)({ case (key, value) => Task((key + key, value + 2)) }).runToFuture.futureValue
+      result shouldBe expected
+    }
+  }
+
+  test("traverseMapValues") {
+    forAll { data: List[(String, Int)] =>
+      val map = data.toMap
+      val expected = map.view.mapValues(value => value + 2).toMap
+      val result = Task.traverseMapValues(map)({ case (key, value) => Task(value + 2) }).runToFuture.futureValue
+      result shouldBe expected
+    }
+  }
+}