Type Members

1.  sealed abstract class Coeval[+A] extends () ⇒ A with Serializable

   Coeval represents lazy computations that can execute synchronously.

   Coeval represents lazy computations that can execute synchronously.

   Word definition and origin:

   • Having the same age or date of origin; a contemporary; synchronous.
   • From the Latin "coævus": com- ‎("equal") in combination with aevum ‎(aevum, "age").
   • The constructor of Coeval is the dual of an expression that evaluates to an A.

   There are three evaluation strategies:

   • now or raiseError: for describing strict values, evaluated immediately
   • evalOnce: expressions evaluated a single time
   • eval: expressions evaluated every time the value is needed

   The Once and Always are both lazy strategies while Now and Error are eager. Once and Always are distinguished from each other only by memoization: once evaluated Once will save the value to be returned immediately if it is needed again. Always will run its computation every time.

   Both Now and Error are represented by the Eager trait, a sub-type of Coeval that can be used as a replacement for Scala's own Try type.

   Coeval supports stack-safe lazy computation via the .map and .flatMap methods, which use an internal trampoline to avoid stack overflows. Computations done within .map and .flatMap are always lazy, even when applied to a Coeval.Eager instance (e.g. Coeval.Now, Coeval.Error).

   Evaluation Strategies

   The "now" and "raiseError" builders are building Coeval instances out of strict values:

   val fa = Coeval.now(1)
   fa.value() // => 1
   
   val fe = Coeval.raiseError(new RuntimeException("dummy"))
   fe.failed // => has RuntimeException

   The "always" strategy is equivalent with a plain function:

   // For didactic purposes, don't use shared vars at home :-)
   var i = 0
   val coeval = Coeval.eval { i += 1; i }
   
   coeval.value() // => 1
   coeval.value() // => 2
   coeval.value() // => 3

   The "once" strategy is equivalent with Scala's lazy val (along with thread-safe idempotency guarantees):

   var j = 0
   val coevalOnce = Coeval.evalOnce { j += 1; j }
   
   coevalOnce.value() // => 1
   coevalOnce.value() // => 1
   coevalOnce.value() // => 1

   Versus Task

   The other option of suspending side-effects is Task. As a quick comparison:

   • Coeval's execution is always immediate / synchronous, whereas Task can describe asynchronous computations
   • Coeval is not cancelable, obviously, since execution is immediate and there's nothing to cancel

   Versus cats.Eval

   The Coeval data type is very similar with cats.Eval. As a quick comparison:

   • cats.Eval is only for controlling laziness, but it doesn't handle side effects, hence cats.Eval is a Comonad
   • Monix's Coeval can handle side effects as well and thus it implements MonadError[Coeval, Throwable] and cats.effect.Sync, providing error-handling utilities

   If you just want to delay the evaluation of a pure expression use cats.Eval, but if you need to suspend side effects or you need error handling capabilities, then use Coeval.

2.  trait CoevalLift[F[_]] extends ~>[Coeval, F]

   A lawless type class that specifies conversions from Coeval to similar data types (i.e.

   A lawless type class that specifies conversions from Coeval to similar data types (i.e. pure, synchronous).

   This is nothing more than a cats.arrow.FunctionK.

   Annotations

   @implicitNotFound( ... )

3.  trait CoevalLike[F[_]] extends ~>[F, Coeval]

   A lawless type class that provides conversions to Coeval.

   A lawless type class that provides conversions to Coeval.

   Sample:

   // Conversion from cats.Eval
   import cats.Eval
   
   val source0 = Eval.always(1 + 1)
   val task0 = CoevalLike[Eval].apply(source0)
   
   // Conversion from SyncIO
   import cats.effect.SyncIO
   
   val source1 = SyncIO(1 + 1)
   val task1 = CoevalLike[SyncIO].apply(source1)

   This is an alternative to usage of cats.effect.Effect where the internals are specialized to Coeval anyway, like for example the implementation of monix.reactive.Observable.

4.  trait Fiber[A] extends cats.effect.Fiber[Task, A]

   Fiber represents the (pure) result of a Task being started concurrently and that can be either joined or cancelled.

   Fiber represents the (pure) result of a Task being started concurrently and that can be either joined or cancelled.

   You can think of fibers as being lightweight threads, a fiber being a concurrency primitive for doing cooperative multi-tasking.

   For example a Fiber value is the result of evaluating Task.start:

   val task = Task.evalAsync(println("Hello!"))
   
   val forked: Task[Fiber[Unit]] = task.start

   Usage example:

   val launchMissiles = Task(println("Missiles launched!"))
   val runToBunker = Task(println("Run Lola run!"))
   
   for {
     fiber <- launchMissiles.start
     _ <- runToBunker.onErrorHandleWith { error =>
       // Retreat failed, cancel launch (maybe we should
       // have retreated to our bunker before the launch?)
       fiber.cancel.flatMap(_ => Task.raiseError(error))
     }
     aftermath <- fiber.join
   } yield {
     aftermath
   }

5.  sealed abstract class Task[+A] extends Serializable

   Task represents a specification for a possibly lazy or asynchronous computation, which when executed will produce an A as a result, along with possible side-effects.

   Task represents a specification for a possibly lazy or asynchronous computation, which when executed will produce an A as a result, along with possible side-effects.

   Compared with Future from Scala's standard library, Task does not represent a running computation or a value detached from time, as Task does not execute anything when working with its builders or operators and it does not submit any work into any thread-pool, the execution eventually taking place only after runAsync is called and not before that.

   Note that Task is conservative in how it spawns logical threads. Transformations like map and flatMap for example will default to being executed on the logical thread on which the asynchronous computation was started. But one shouldn't make assumptions about how things will end up executed, as ultimately it is the implementation's job to decide on the best execution model. All you are guaranteed is asynchronous execution after executing runAsync.

   Getting Started

   To build a Task from a by-name parameters (thunks), we can use Task.apply ( alias Task.eval) or Task.evalAsync:

   val hello = Task("Hello ")
   val world = Task.evalAsync("World!")

   Nothing gets executed yet, as Task is lazy, nothing executes until you trigger its evaluation via runAsync or runToFuture.

   To combine Task values we can use .map and .flatMap, which describe sequencing and this time it's in a very real sense because of the laziness involved:

   val sayHello = hello
     .flatMap(h => world.map(w => h + w))
     .map(println)

   This Task reference will trigger a side effect on evaluation, but not yet. To make the above print its message:

   import monix.execution.CancelableFuture
   import monix.execution.Scheduler.Implicits.global
   
   val f: CancelableFuture[Unit] = sayHello.runToFuture
   // => Hello World!

   The returned type is a CancelableFuture which inherits from Scala's standard Future, a value that can be completed already or might be completed at some point in the future, once the running asynchronous process finishes. Such a future value can also be canceled, see below.

   Laziness, Purity and Referential Transparency

   The fact that Task is lazy whereas Future is not has real consequences. For example with Task you can do this:

   import scala.concurrent.duration._
   
   def retryOnFailure[A](times: Int, source: Task[A]): Task[A] =
     source.onErrorHandleWith { err =>
       // No more retries left? Re-throw error:
       if (times <= 0) Task.raiseError(err) else {
         // Recursive call, yes we can!
         retryOnFailure(times - 1, source)
           // Adding 500 ms delay for good measure
           .delayExecution(500.millis)
       }
     }

   Future being a strict value-wannabe means that the actual value gets "memoized" (means cached), however Task is basically a function that can be repeated for as many times as you want.

   Task is a pure data structure that can be used to describe pure functions, the equivalent of Haskell's IO.

   Memoization

   Task can also do memoization, making it behave like a "lazy" Scala Future, meaning that nothing is started yet, its side effects being evaluated on the first runAsync and then the result reused on subsequent evaluations:

   Task(println("boo")).memoize

   The difference between this and just calling runAsync() is that memoize() still returns a Task and the actual memoization happens on the first runAsync() (with idempotency guarantees of course).

   But here's something else that the Future data type cannot do, memoizeOnSuccess:

   Task.eval {
     if (scala.util.Random.nextDouble() > 0.33)
       throw new RuntimeException("error!")
     println("moo")
   }.memoizeOnSuccess

   This keeps repeating the computation for as long as the result is a failure and caches it only on success. Yes we can!

   WARNING: as awesome as memoize can be, use with care because memoization can break referential transparency!

   Parallelism

   Because of laziness, invoking Task.sequence will not work like it does for Future.sequence, the given Task values being evaluated one after another, in sequence, not in parallel. If you want parallelism, then you need to use Task.gather and thus be explicit about it.

   This is great because it gives you the possibility of fine tuning the execution. For example, say you want to execute things in parallel, but with a maximum limit of 30 tasks being executed in parallel. One way of doing that is to process your list in batches:

   // Some array of tasks, you come up with something good :-)
   val list: Seq[Task[Int]] = Seq.tabulate(100)(Task(_))
   
   // Split our list in chunks of 30 items per chunk,
   // this being the maximum parallelism allowed
   val chunks = list.sliding(30, 30).toSeq
   
   // Specify that each batch should process stuff in parallel
   val batchedTasks = chunks.map(chunk => Task.gather(chunk))
   // Sequence the batches
   val allBatches = Task.sequence(batchedTasks)
   
   // Flatten the result, within the context of Task
   val all: Task[Seq[Int]] = allBatches.map(_.flatten)

   Note that the built Task reference is just a specification at this point, or you can view it as a function, as nothing has executed yet, you need to call runAsync or runToFuture explicitly.

   Cancellation

   The logic described by an Task task could be cancelable, depending on how the Task gets built.

   CancelableFuture references can also be canceled, in case the described computation can be canceled. When describing Task tasks with Task.eval nothing can be cancelled, since there's nothing about a plain function that you can cancel, but we can build cancelable tasks with Task.cancelable.

   import scala.concurrent.duration._
   import scala.util._
   
   val delayedHello = Task.cancelable0[Unit] { (scheduler, callback) =>
     val task = scheduler.scheduleOnce(1.second) {
       println("Delayed Hello!")
       // Signaling successful completion
       callback(Success(()))
     }
     // Returning a cancel token that knows how to cancel the
     // scheduled computation:
     Task {
       println("Cancelling!")
       task.cancel()
     }
   }

   The sample above prints a message with a delay, where the delay itself is scheduled with the injected Scheduler. The Scheduler is in fact an implicit parameter to runAsync().

   This action can be cancelled, because it specifies cancellation logic. In case we have no cancelable logic to express, then it's OK if we returned a Cancelable.empty reference, in which case the resulting Task would not be cancelable.

   But the Task we just described is cancelable, for one at the edge, due to runAsync returning Cancelable and CancelableFuture references:

   // Triggering execution
   val cf: CancelableFuture[Unit] = delayedHello.runToFuture
   
   // If we change our mind before the timespan has passed:
   cf.cancel()

   But also cancellation is described on Task as a pure action, which can be used for example in race conditions:

   import scala.concurrent.duration._
   import scala.concurrent.TimeoutException
   
   val ta = Task(1 + 1).delayExecution(4.seconds)
   
   val tb = Task.raiseError[Int](new TimeoutException)
     .delayExecution(4.seconds)
   
   Task.racePair(ta, tb).flatMap {
     case Left((a, fiberB)) =>
       fiberB.cancel.map(_ => a)
     case Right((fiberA, b)) =>
       fiberA.cancel.map(_ => b)
   }

   The returned type in racePair is Fiber, which is a data type that's meant to wrap tasks linked to an active process and that can be canceled or joined.

   Also, given a task, we can specify actions that need to be triggered in case of cancellation, see doOnCancel:

   val task = Task.eval(println("Hello!")).executeAsync
   
   task doOnCancel Task.eval {
     println("A cancellation attempt was made!")
   }

   Given a task, we can also create a new task from it that atomic (non cancelable), in the sense that either all of it executes or nothing at all, via uncancelable.

   Note on the ExecutionModel

   Task is conservative in how it introduces async boundaries. Transformations like map and flatMap for example will default to being executed on the current call stack on which the asynchronous computation was started. But one shouldn't make assumptions about how things will end up executed, as ultimately it is the implementation's job to decide on the best execution model. All you are guaranteed (and can assume) is asynchronous execution after executing runAsync.

   Currently the default ExecutionModel specifies batched execution by default and Task in its evaluation respects the injected ExecutionModel. If you want a different behavior, you need to execute the Task reference with a different scheduler.

6.  trait TaskApp extends AnyRef

   Safe App type that executes a Task.

   Safe App type that executes a Task. Shutdown occurs after the Task completes, as follows:

   - If completed with ExitCode.Success, the main method exits and shutdown is handled by the platform.

   - If completed with any other ExitCode, sys.exit is called with the specified code.

   - If the Task raises an error, the stack trace is printed to standard error and sys.exit(1) is called.

   When a shutdown is requested via a signal, the Task is canceled and we wait for the IO to release any resources. The process exits with the numeric value of the signal plus 128.

   import cats.effect._
   import cats.implicits._
   import monix.eval._
   
   object MyApp extends TaskApp {
     def run(args: List[String]): Task[ExitCode] =
       args.headOption match {
         case Some(name) =>
           Task(println(s"Hello, \${name}.")).as(ExitCode.Success)
         case None =>
           Task(System.err.println("Usage: MyApp name")).as(ExitCode(2))
       }
   }

   N.B. this is homologous with cats.effect.IOApp, but meant for usage with Task.

   Works on top of JavaScript as well ;-)

7.  trait TaskLift[F[_]] extends ~>[Task, F]

   A lawless type class that specifies conversions from Task to similar data types (i.e.

   A lawless type class that specifies conversions from Task to similar data types (i.e. pure, asynchronous, preferably cancelable).

   Annotations

   @implicitNotFound( ... )

8.  trait TaskLike[F[_]] extends ~>[F, Task]

   A lawless type class that provides conversions to Task.

   A lawless type class that provides conversions to Task.

   Sample:

   // Conversion from cats.Eval
   import cats.Eval
   
   val source0 = Eval.always(1 + 1)
   val task0 = TaskLike[Eval].apply(source0)
   
   // Conversion from Future
   import scala.concurrent.Future
   
   val source1 = Future.successful(1 + 1)
   val task1 = TaskLike[Future].apply(source1)
   
   // Conversion from IO
   import cats.effect.IO
   
   val source2 = IO(1 + 1)
   val task2 = TaskLike[IO].apply(source2)

   This is an alternative to usage of cats.effect.Effect where the internals are specialized to Task anyway, like for example the implementation of monix.reactive.Observable.

   Annotations

   @implicitNotFound( ... )

9.  final class TaskLocal[A] extends AnyRef

   A TaskLocal is like a ThreadLocal that is pure and with a flexible scope, being processed in the context of the Task data type.

   A TaskLocal is like a ThreadLocal that is pure and with a flexible scope, being processed in the context of the Task data type.

   This data type wraps monix.execution.misc.Local.

   Just like a ThreadLocal, usage of a TaskLocal is safe, the state of all current locals being transported over async boundaries (aka when threads get forked) by the Task run-loop implementation, but only when the Task reference gets executed with Task.Options.localContextPropagation set to true.

   One way to achieve this is with Task.executeWithOptions, a single call is sufficient just before runAsync:

   import monix.execution.Scheduler.Implicits.global
   
   val t = Task(42)
   t.executeWithOptions(_.enableLocalContextPropagation)
     // triggers the actual execution
     .runToFuture

   Another possibility is to use Task.runToFutureOpt or Task.runToFutureOpt instead of runAsync and specify the set of options implicitly:

   {
     implicit val options = Task.defaultOptions.enableLocalContextPropagation
   
     // Options passed implicitly
     val f = t.runToFutureOpt
   }

   Full example:

   import monix.eval.{Task, TaskLocal}
   
   val task: Task[Unit] =
     for {
       local <- TaskLocal(0)
       value1 <- local.read // value1 == 0
       _ <- local.write(100)
       value2 <- local.read // value2 == 100
       value3 <- local.bind(200)(local.read.map(_ * 2)) // value3 == 200 * 2
       value4 <- local.read // value4 == 100
       _ <- local.clear
       value5 <- local.read // value5 == 0
     } yield {
       // Should print 0, 100, 400, 100, 0
       println("value1: " + value1)
       println("value2: " + value2)
       println("value3: " + value3)
       println("value4: " + value4)
       println("value5: " + value5)
     }
   
   // For transporting locals over async boundaries defined by
   // Task, any Scheduler will do, however for transporting locals
   // over async boundaries managed by Future and others, you need
   // a `TracingScheduler` here:
   import monix.execution.Scheduler.Implicits.global
   
   // Needs enabling the "localContextPropagation" option
   // just before execution
   implicit val opts = Task.defaultOptions.enableLocalContextPropagation
   
   // Triggering actual execution
   val result = task.runToFutureOpt

10.  type MVar[A] = catnap.MVar[Task, A]

    DEPRECATED — moved and made generic in monix.catnap.CircuitBreaker.

    DEPRECATED — moved and made generic in monix.catnap.CircuitBreaker.

    Please switch to that, because the deprecated symbols will be removed.

    Annotations

    @deprecated

    Deprecated

    (Since version 3.0.0) Moved and made generic in monix.catnap.MVar

11.  type TaskCircuitBreaker = CircuitBreaker[Task]

    DEPRECATED — moved and made generic in monix.catnap.CircuitBreaker.

    DEPRECATED — moved and made generic in monix.catnap.CircuitBreaker.

    Please switch to that, because the deprecated symbols will be removed.

    Annotations

    @deprecated

    Deprecated

    (Since version 3.0.0) Moved and made generic in monix.catnap.CircuitBreaker

12.  type TaskSemaphore = Semaphore[Task]

    DEPRECATED — moved and made generic in monix.catnap.Semaphore.

    DEPRECATED — moved and made generic in monix.catnap.Semaphore.

    Please switch to that, because the deprecated symbols will be removed.

    Annotations

    @deprecated

    Deprecated

    (Since version 3.0.0) Moved and made generic in monix.catnap.Semaphore