Re-implement using callbacks. (#188)

* Re-implement using callbacks.

Advantages:

* The ZIO → Monix conversion is now lazy
* Interruption/cancellation can be propagated
* Executes synchronously where possible

Also removed the need for the user to provide a Monix scheduler,
instead we create one from the ZIO runtime.

* Fix tests.

* Apply suggestions from code review

Co-authored-by: Dejan Mijić <dmijic@acm.org>

README.md

@@ -1,92 +1,87 @@
-# Interop Monix
+# Monix Interoperability for ZIO
 
 | Project Stage | CI | Release | Snapshot | Discord |
 | --- | --- | --- | --- | --- |
 | [![Project stage][Stage]][Stage-Page] | ![CI][Badge-CI] | [![Release Artifacts][Badge-SonatypeReleases]][Link-SonatypeReleases] | [![Snapshot Artifacts][Badge-SonatypeSnapshots]][Link-SonatypeSnapshots] | [![Discord][Badge-Discord]][Link-Discord] |
 
-## Task conversions
+This library provides interoperability between **Monix 3.4** and **ZIO 1 and ZIO 2**.
 
-Interop layer provides the following conversions:
+## Tasks
 
-- from `Task[A]` to `UIO[Task[A]]`
-- from `Task[A]` to `Task[A]`
-
-To convert an `IO` value to `Task`, use the following method:
+Monix tasks can be converted to ZIO tasks:
 
 ```scala
-def toTask: UIO[eval.Task[A]]
-```
+import zio._
+import zio.interop.monix._
+import monix.eval
 
-To perform conversion in other direction, use the following extension method
-available on `IO` companion object:
+val monixTask: eval.Task[String] = ???
 
-```scala
-def fromTask[A](task: eval.Task[A])(implicit scheduler: Scheduler): Task[A]
+val zioTask: Task[String] = ZIO.fromMonixTask(monixTask)
 ```
 
-Note that in order to convert the `Task` to an `IO`, an appropriate `Scheduler`
-needs to be available.
+The conversion is lazy: the Monix task will only be executed if the returned ZIO task is executed.
 
-### Example
+ZIO tasks can be converted to Monix tasks:
 
 ```scala
-import monix.eval.Task
-import monix.execution.Scheduler.Implicits.global
-import zio.{ IO, DefaultRuntime }
+import zio._
 import zio.interop.monix._
+import monix.eval
+import monix.execution.Scheduler.Implicits.global
 
-object UnsafeExample extends DefaultRuntime {
-  def main(args: Array[String]): Unit = {
-    val io1 = IO.succeed(10)
-    val t1  = unsafeRun(io1.toTask)
-
-    t1.runToFuture.foreach(r => println(s"IO to task result is $r"))
+val zioTask: Task[String] = ???
 
-    val t2  = Task(10)
-    val io2 = IO.fromTask(t2).map(r => s"Task to IO result is $r")
+val createMonixTask: UIO[eval.Task[String]] = zioTask.toMonixTask()
 
-    println(unsafeRun(io2))
-  }
-}
+// illustrative, you wouldn't usually do things this way
+val monixTask: eval.Task[String] = Runtime.default.unsafeRun(createMonixTask)
+val stringResult = monixTask.runSyncUnsafe
 ```
 
-## Coeval conversions
+The conversion is lazy: the ZIO effect so converted will only be executed if the returned Monix task is executed.
 
-To convert an `IO` value to `Coeval`, use the following method:
+Sometimes you need to provide a Monix task in a context where using a ZIO effect is difficult. For example, when an API requires you to provide a function that returns a Monix task. In these situations, the `toMonixTaskUsingRuntime` method can be used:
 
 ```scala
-def toCoeval: UIO[eval.Coeval[A]]
-```
+import zio._
+import zio.interop.monix._
+import monix.eval
 
-To perform conversion in other direction, use the following extension method
-available on `IO` companion object:
+def monixBasedApi(f: String => eval.Task[Unit]): eval.Task[Unit] = ???
 
-```scala
-def fromCoeval[A](coeval: eval.Coeval[A]): Task[A]
+def zioBasedProcessor(s: String): Task[Unit] = ???
+
+val zioEffects = for {
+    zioRuntime <- ZIO.runtime[Any]
+    monixTask = 
+    _ <- ZIO.fromMonixTask {
+        monixBasedApi(s =>
+            zioBasedProcessor(s).toMonixTaskUsingRuntime(zioRuntime)
+        )
+    }
+} yield ()
 ```
 
-### Example
+Cancellation/Interruption is propagated between the effect systems. Interrupting a ZIO task based on a Monix task will cancel the underlying Monix task and vice-versa. Be aware that ZIO interruption does not return until cancellation effects have completed, whereas Monix cancellation returns as soon as the signal is sent, without waiting for the cancellation effects to complete.
 
-```scala
-import monix.eval.Coeval
-import zio.{ IO, DefaultRuntime }
-import zio.interop.monix._
+## Monix Scheduler
 
-object UnsafeExample extends DefaultRuntime {
-  def main(args: Array[String]): Unit = {
-    val io1 = IO.succeed(10)
-    val c1  = unsafeRun(io1.toCoeval) 
+Sometimes it is useful to have a Monix `Scheduler` available for interop purposes. The `Runtime#monixScheduler` method will create a scheduler that shares its execution context with the ZIO runtime:
 
-    println(s"IO to coeval result is ${c1.value}")
+```scala
+import zio._
+import zio.interop.monix._
+import monix.execution.Scheduler
 
-    val c2  = Coeval(10)
-    val io2 = IO.fromCoeval(c2).map(r => s"Coeval to IO result is $r")
+ZIO.runtime[Any].flatMap { runtime =>
+    implicit val monixScheduler: Scheduler = runtime.monixScheduler()
 
-    println(unsafeRun(io2))
-  }
+    // do Monixy things
 }
 ```
 
+
 [Badge-CI]: https://github.com/zio/interop-monix/workflows/CI/badge.svg
 [Badge-Discord]: https://img.shields.io/discord/629491597070827530?logo=discord
 [Badge-SonatypeReleases]: https://img.shields.io/nexus/r/https/oss.sonatype.org/dev.zio/zio-interop-monix_2.12.svg

build.sbt

@@ -44,7 +44,7 @@ lazy val interopMonix = crossProject(JSPlatform, JVMPlatform)
     libraryDependencies ++= Seq(
       "io.monix" %%% "monix"        % "3.4.0",
       "dev.zio"  %%% "zio"          % "1.0.13",
-      "dev.zio"  %%% "zio-test"     % "1.0.13",
+      "dev.zio"  %%% "zio-test"     % "1.0.13" % Test,
       "dev.zio"  %%% "zio-test-sbt" % "1.0.13" % Test
     )
   )

interop-monix/shared/src/main/scala/zio/interop/monix/package.scala

@@ -0,0 +1,109 @@
+package zio
+package interop
+
+import _root_.monix.eval.{ Task => MTask }
+import _root_.monix.execution.{ Scheduler => MScheduler, ExecutionModel }
+
+/**
+ * Monix interoperability for ZIO.
+ *
+ *   - Monix tasks can be converted to ZIO tasks via
+ *     `ZIO.fromMonixTask(monixTask)`.
+ *   - ZIO tasks can be converted to Monix tasks via `zioTask.toMonixTask`.
+ */
+package object monix {
+
+  implicit final class ZIORuntimeOps[R](private val runtime: Runtime[R]) extends AnyVal {
+
+    /**
+     * Creates a Monix scheduler that shares its execution context with this ZIO
+     * runtime.
+     */
+    def monixScheduler(executionModel: ExecutionModel = ExecutionModel.Default): MScheduler =
+      MScheduler(runtime.platform.executor.asEC, executionModel)
+
+  }
+
+  implicit final class ZIOObjOps(private val unused: ZIO.type) extends AnyVal {
+
+    /**
+     * Converts a Monix task into a ZIO task.
+     *
+     * Interrupting the returned effect will cancel the underlying Monix task.
+     * The conversion is lazy: the Monix task is only executed if the returned
+     * ZIO task is executed.
+     *
+     * If the returned ZIO task is interrupted, the underlying Monix task will be
+     * cancelled.
+     *
+     * @param monixTask
+     *   The Monix task.
+     * @param executionModel
+     *   The Monix execution model to use for the Monix execution. This only
+     *   need be specified if you want to override the Monix default.
+     */
+    def fromMonixTask[A](monixTask: MTask[A], executionModel: ExecutionModel = ExecutionModel.Default): Task[A] =
+      Task.runtime.flatMap { zioRuntime =>
+        Task.effectAsyncInterrupt[A] { cb =>
+          implicit val scheduler: MScheduler = zioRuntime.monixScheduler(executionModel)
+          try
+          // runSyncStep will try to execute the Monix effects synchronously
+          // if it fails before hitting an async boundary, the failure will be thrown
+          monixTask.runSyncStep match {
+            case Right(result)        =>
+              // Monix task ran synchronously and successfully
+              Right(ZIO.succeedNow(result))
+            case Left(asyncMonixTask) =>
+              // Monix task hit an async boundary, so we have to use the callback (cb)
+              // and return a cancelation effect
+              val cancelable = asyncMonixTask.runAsync { result =>
+                val zioEffect = ZIO.fromEither(result)
+                cb(zioEffect)
+              }
+              Left(ZIO.succeed(cancelable.cancel()))
+          } catch {
+            // Monix task failed during synchronous execution
+            case e: Throwable => Right(ZIO.fail(e))
+          }
+        }
+      }
+
+  }
+
+  implicit final class ExtraZioEffectOps[-R, +A](private val effect: ZIO[R, Throwable, A]) extends AnyVal {
+
+    /**
+     * Converts this ZIO effect into a Monix task.
+     *
+     * The conversion is lazy: this effect will only be executed if the returned Monix task
+     * is executed.
+     * If the returned Monix task is cancelled, the underlying ZIO effect will
+     * be interrupted.
+     */
+    def toMonixTask: URIO[R, MTask[A]] = ZIO.runtime[R].map(toMonixTaskUsingRuntime)
+
+    /**
+     * Converts this ZIO effect into a Monix task using a specified ZIO runtime
+     * to run the ZIO effect.
+     *
+     * This is useful in situations where a Monix task is needed, but executing ZIO effects
+     * is incovenient. For example, when using a library API that requires you to pass a
+     * function that returns a Monix task. In such situations, you can acquire the ZIO
+     * runtime up-front and use this method to create the Monix task.
+     *
+     * The conversion is lazy: this effect will only be executed if the returned Monix task
+     * is executed.
+     * If the returned Monix task is cancelled, the underlying ZIO effect will
+     * be interrupted.
+     */
+    def toMonixTaskUsingRuntime(zioRuntime: Runtime[R]): MTask[A] =
+      MTask.cancelable { cb =>
+        val cancelable = zioRuntime.unsafeRunAsyncCancelable(effect) { exit =>
+          exit.fold(failed => cb.onError(failed.squash), cb.onSuccess)
+        }
+        MTask.eval(cancelable(Fiber.Id.None)).void
+      }
+
+  }
+
+}