Runs this task first and then, when successful, the given task. Returns the result of the given task.

Example:

val combined = Task{println("first"); "first"} >> Task{println("second"); "second"}
// Prints "first" and then "second"
// Result value will be "second"

Introduces an asynchronous boundary at the current stage in the asynchronous processing pipeline, making processing to jump on the given Scheduler (until the next async boundary).

Consider the following example:

import monix.execution.Scheduler
val io = Scheduler.io()

val source = Task(1).executeOn(io).map(_ + 1)

That task is being forced to execute on the io scheduler, including the map transformation that follows after executeOn. But what if we want to jump with the execution run-loop on another scheduler for the following transformations?

Then we can do:

import monix.execution.Scheduler.global

source.asyncBoundary(global).map(_ + 2)

In this sample, whatever gets evaluated by the source will happen on the io scheduler, however the asyncBoundary call will make all subsequent operations to happen on the specified global scheduler.

Introduces an asynchronous boundary at the current stage in the asynchronous processing pipeline.

Consider the following example:

import monix.execution.Scheduler
val io = Scheduler.io()

val source = Task(1).executeOn(io).map(_ + 1)

That task is being forced to execute on the io scheduler, including the map transformation that follows after executeOn. But what if we want to jump with the execution run-loop on the default scheduler for the following transformations?

Then we can do:

source.asyncBoundary.map(_ + 2)

In this sample, whatever gets evaluated by the source will happen on the io scheduler, however the asyncBoundary call will make all subsequent operations to happen on the default scheduler.

Returns a task that treats the source task as the acquisition of a resource, which is then exploited by the use function and then released.

The bracket operation is the equivalent of the try {} catch {} finally {} statements from mainstream languages.

The bracket operation installs the necessary exception handler to release the resource in the event of an exception being raised during the computation, or in case of cancellation.

If an exception is raised, then bracket will re-raise the exception after performing the release. If the resulting task gets cancelled, then bracket will still perform the release, but the yielded task will be non-terminating (equivalent with Task.never).

Example:

import java.io._

def readFile(file: File): Task[String] = {
  // Opening a file handle for reading text
  val acquire = Task.eval(new BufferedReader(
    new InputStreamReader(new FileInputStream(file), "utf-8")
  ))

  acquire.bracket { in =>
    // Usage part
    Task.eval {
      // Yes, ugly Java, non-FP loop;
      // side-effects are suspended though
      var line: String = null
      val buff = new StringBuilder()
      do {
        line = in.readLine()
        if (line != null) buff.append(line)
      } while (line != null)
      buff.toString()
    }
  } { in =>
    // The release part
    Task.eval(in.close())
  }
}

Note that in case of cancellation the underlying implementation cannot guarantee that the computation described by use doesn't end up executed concurrently with the computation from release. In the example above that ugly Java loop might end up reading from a BufferedReader that is already closed due to the task being cancelled, thus triggering an error in the background with nowhere to go but in Scheduler.reportFailure.

In this particular example, given that we are just reading from a file, it doesn't matter. But in other cases it might matter, as concurrency on top of the JVM when dealing with I/O might lead to corrupted data.

For those cases you might want to do synchronization (e.g. usage of locks and semaphores) and you might want to use bracketE, the version that allows you to differentiate between normal termination and cancellation.

NOTE on error handling: one big difference versus try {} finally {} is that, in case both the release function and the use function throws, the error raised by use gets signaled and the error raised by release gets reported with System.err for Coeval or with Scheduler.reportFailure for Task.

For example:

Task.evalAsync("resource").bracket { _ =>
  // use
  Task.raiseError(new RuntimeException("Foo"))
} { _ =>
  // release
  Task.raiseError(new RuntimeException("Bar"))
}

In this case the error signaled downstream is "Foo", while the "Bar" error gets reported. This is consistent with the behavior of Haskell's bracket operation and NOT with try {} finally {} from Scala, Java or JavaScript.

Returns a new task that treats the source task as the acquisition of a resource, which is then exploited by the use function and then released, with the possibility of distinguishing between normal termination and cancelation, such that an appropriate release of resources can be executed.

The bracketCase operation is the equivalent of try {} catch {} finally {} statements from mainstream languages when used for the acquisition and release of resources.

The bracketCase operation installs the necessary exception handler to release the resource in the event of an exception being raised during the computation, or in case of cancelation.

In comparison with the simpler bracket version, this one allows the caller to differentiate between normal termination, termination in error and cancelation via an ExitCase parameter.

Returns a task that treats the source task as the acquisition of a resource, which is then exploited by the use function and then released, with the possibility of distinguishing between normal termination and cancellation, such that an appropriate release of resources can be executed.

The bracketE operation is the equivalent of try {} catch {} finally {} statements from mainstream languages.

The bracketE operation installs the necessary exception handler to release the resource in the event of an exception being raised during the computation, or in case of cancellation.

In comparison with the simpler bracket version, this one allows the caller to differentiate between normal termination and cancellation.

The release function receives as input:

• Left(None) in case of cancellation
• Left(Some(error)) in case use terminated with an error
• Right(b) in case of success

NOTE on error handling: one big difference versus try {} finally {} is that, in case both the release function and the use function throws, the error raised by use gets signaled and the error raised by release gets reported with System.err for Coeval or with Scheduler.reportFailure for Task.

For example:

Task.evalAsync("resource").bracket { _ =>
  // use
  Task.raiseError(new RuntimeException("Foo"))
} { _ =>
  // release
  Task.raiseError(new RuntimeException("Bar"))
}

In this case the error signaled downstream is "Foo", while the "Bar" error gets reported. This is consistent with the behavior of Haskell's bracket operation and NOT with try {} finally {} from Scala, Java or JavaScript.

Returns a task that waits for the specified timespan before executing and mirroring the result of the source.

In this example we're printing to standard output, but before doing that we're introducing a 3 seconds delay:

import scala.concurrent.duration._

Task(println("Hello!"))
  .delayExecution(3.seconds)

This operation is also equivalent with:

Task.sleep(3.seconds).flatMap(_ => Task(println("Hello!")))

See Task.sleep for the operation that describes the effect and Task.delayResult for the version that evaluates the task on time, but delays the signaling of the result.

Returns a task that executes the source immediately on runAsync, but before emitting the onSuccess result for the specified duration.

Note that if an error happens, then it is streamed immediately with no delay.

See delayExecution for delaying the evaluation of the task with the specified duration. The delayResult operation is effectively equivalent with:

import scala.concurrent.duration._

Task(1 + 1)
  .flatMap(a => Task.now(a).delayExecution(3.seconds))

Or if we are to use the Task.sleep describing just the effect, this operation is equivalent with:

Task(1 + 1).flatMap(a => Task.sleep(3.seconds).map(_ => a))

Thus in this example 3 seconds will pass before the result is being generated by the source, plus another 5 seconds before it is finally emitted:

Task(1 + 1)
  .delayExecution(3.seconds)
  .delayResult(5.seconds)

Returns a new Task that will mirror the source, but that will execute the given callback if the task gets canceled before completion.

This only works for premature cancellation. See doOnFinish for triggering callbacks when the source finishes.

Returns a new Task in which f is scheduled to be run on completion. This would typically be used to release any resources acquired by this Task.

The returned Task completes when both the source and the task returned by f complete.

NOTE: The given function is only called when the task is complete. However the function does not get called if the task gets canceled. Cancellation is a process that's concurrent with the execution of a task and hence needs special handling.

See doOnCancel for specifying a callback to call on canceling a task.

Mirrors the given source Task, but upon execution ensure that evaluation forks into a separate (logical) thread.

The Scheduler used will be the one that is used to start the run-loop in Task.runAsync or Task.runToFuture.

This operation is equivalent with:

Task.shift.flatMap(_ => Task(1 + 1))

// ... or ...

import cats.syntax.all._

Task.shift *> Task(1 + 1)

The Scheduler used for scheduling the async boundary will be the default, meaning the one used to start the run-loop in runAsync.

Overrides the default Scheduler, possibly forcing an asynchronous boundary before execution (if forceAsync is set to true, the default).

When a Task is executed with Task.runAsync or Task.runToFuture, it needs a Scheduler, which is going to be injected in all asynchronous tasks processed within the flatMap chain, a Scheduler that is used to manage asynchronous boundaries and delayed execution.

This scheduler passed in runAsync is said to be the "default" and executeOn overrides that default.

import monix.execution.Scheduler
import java.io.{BufferedReader, FileInputStream, InputStreamReader}

/** Reads the contents of a file using blocking I/O. */
def readFile(path: String): Task[String] = Task.eval {
  val in = new BufferedReader(
    new InputStreamReader(new FileInputStream(path), "utf-8"))

  val buffer = new StringBuffer()
  var line: String = null
  do {
    line = in.readLine()
    if (line != null) buffer.append(line)
  } while (line != null)

  buffer.toString
}

// Building a Scheduler meant for blocking I/O
val io = Scheduler.io()

// Building the Task reference, specifying that `io` should be
// injected as the Scheduler for managing async boundaries
readFile("path/to/file").executeOn(io, forceAsync = true)

In this example we are using Task.eval, which executes the given thunk immediately (on the current thread and call stack).

By calling executeOn(io), we are ensuring that the used Scheduler (injected in async tasks) will be io, a Scheduler that we intend to use for blocking I/O actions. And we are also forcing an asynchronous boundary right before execution, by passing the forceAsync parameter as true (which happens to be the default value).

Thus, for our described function that reads files using Java's blocking I/O APIs, we are ensuring that execution is entirely managed by an io scheduler, executing that logic on a thread pool meant for blocking I/O actions.

Note that in case forceAsync = false, then the invocation will not introduce any async boundaries of its own and will not ensure that execution will actually happen on the given Scheduler, that depending of the implementation of the Task. For example:

Task.eval("Hello, " + "World!")
  .executeOn(io, forceAsync = false)

The evaluation of this task will probably happen immediately (depending on the configured ExecutionModel) and the given scheduler will probably not be used at all.

However in case we would use Task.apply, which ensures that execution of the provided thunk will be async, then by using executeOn we'll indeed get a logical fork on the io scheduler:

Task("Hello, " + "World!").executeOn(io, forceAsync = false)

Also note that overriding the "default" scheduler can only happen once, because it's only the "default" that can be overridden.

Something like this won't have the desired effect:

val io1 = Scheduler.io()
val io2 = Scheduler.io()

Task(1 + 1).executeOn(io1).executeOn(io2)

In this example the implementation of task will receive the reference to io1 and will use it on evaluation, while the second invocation of executeOn will create an unnecessary async boundary (if forceAsync = true) or be basically a costly no-op. This might be confusing but consider the equivalence to these functions:

import scala.concurrent.ExecutionContext

val io11 = Scheduler.io()
val io22 = Scheduler.io()

def sayHello(ec: ExecutionContext): Unit =
  ec.execute(new Runnable {
    def run() = println("Hello!")
  })

def sayHello2(ec: ExecutionContext): Unit =
  // Overriding the default `ec`!
  sayHello(io11)

def sayHello3(ec: ExecutionContext): Unit =
  // Overriding the default no longer has the desired effect
  // because sayHello2 is ignoring it!
  sayHello2(io22)

Returns a new task that will execute the source with a different ExecutionModel.

This allows fine-tuning the options injected by the scheduler locally. Example:

import monix.execution.ExecutionModel.AlwaysAsyncExecution
Task(1 + 1).executeWithModel(AlwaysAsyncExecution)

Returns a new task that will execute the source with a different set of Options.

This allows fine-tuning the default options. Example:

Task(1 + 1).executeWithOptions(_.enableAutoCancelableRunLoops)

Returns a failed projection of this task.

The failed projection is a Task holding a value of type Throwable, emitting the error yielded by the source, in case the source fails, otherwise if the source succeeds the result will fail with a NoSuchElementException.

Describes flatMap-driven loops, as an alternative to recursive functions.

Sample:

import scala.util.Random

val random = Task(Random.nextInt())
val loop = random.flatMapLoop(Vector.empty[Int]) { (a, list, continue) =>
  val newList = list :+ a
  if (newList.length < 5)
    continue(newList)
  else
    Task.now(newList)
}

Triggers the evaluation of the source, executing the given function for the generated element.

The application of this function has strict behavior, as the task is immediately executed.

Exceptions in f are reported using provided (implicit) Scheduler

Returns a new task that upon evaluation will execute the given function for the generated element, transforming the source into a Task[Unit].

Similar in spirit with normal foreach, but lazy, as obviously nothing gets executed at this point.

Executes the given finalizer when the source is finished, either in success or in error, or if canceled.

This variant of guaranteeCase evaluates the given finalizer regardless of how the source gets terminated:

• normal completion
• completion in error
• cancellation

As best practice, it's not a good idea to release resources via guaranteeCase in polymorphic code. Prefer bracket for the acquisition and release of resources.

Executes the given finalizer when the source is finished, either in success or in error, or if canceled, allowing for differentiating between exit conditions.

This variant of guarantee injects an ExitCase in the provided function, allowing one to make a difference between:

• normal completion
• completion in error
• cancellation

As best practice, it's not a good idea to release resources via guaranteeCase in polymorphic code. Prefer bracketCase for the acquisition and release of resources.

Returns a new Task that repeatedly executes the source as long as it continues to succeed. It never produces a terminal value.

Example:

import scala.concurrent.duration._

Task.eval(println("Tick!"))
  .delayExecution(1.second)
  .loopForever

Returns a new Task that applies the mapping function to the element emitted by the source.

Can be used for specifying a (lazy) transformation to the result of the source.

This equivalence with flatMap always holds:

fa.map(f) <-> fa.flatMap(x => Task.pure(f(x)))

Memoizes (caches) the result of the source task and reuses it on subsequent invocations of runAsync.

The resulting task will be idempotent, meaning that evaluating the resulting task multiple times will have the same effect as evaluating it once.

Cancellation — a memoized task will mirror the behavior of the source on cancellation. This means that:

• if the source isn't cancellable, then the resulting memoized task won't be cancellable either
• if the source is cancellable, then the memoized task can be cancelled, which can take unprepared users by surprise

Depending on use-case, there are two ways to ensure no surprises:

• usage of onCancelRaiseError, before applying memoization, to ensure that on cancellation an error is triggered and then noticed by the memoization logic
• usage of uncancelable, either before or after applying memoization, to ensure that the memoized task cannot be cancelled

Example:

import scala.concurrent.CancellationException
import scala.concurrent.duration._

val source = Task(1).delayExecution(5.seconds)

// Option 1: trigger error on cancellation
val err = new CancellationException
val cached1 = source.onCancelRaiseError(err).memoize

// Option 2: make it uninterruptible
val cached2 = source.uncancelable.memoize

When using onCancelRaiseError like in the example above, the behavior of memoize is to cache the error. If you want the ability to retry errors until a successful value happens, see memoizeOnSuccess.

UNSAFE — this operation allocates a shared, mutable reference, which can break in certain cases referential transparency, even if this operation guarantees idempotency (i.e. referential transparency implies idempotency, but idempotency does not imply referential transparency).

The allocation of a mutable reference is known to be a side effect, thus breaking referential transparency, even if calling this method does not trigger the evaluation of side effects suspended by the source.

Use with care. Sometimes it's easier to just keep a shared, memoized reference to some connection, but keep in mind it might be better to pass such a reference around as a parameter.

Memoizes (cache) the successful result of the source task and reuses it on subsequent invocations of runAsync. Thrown exceptions are not cached.

The resulting task will be idempotent, but only if the result is successful.

Cancellation — a memoized task will mirror the behavior of the source on cancellation. This means that:

• if the source isn't cancellable, then the resulting memoized task won't be cancellable either
• if the source is cancellable, then the memoized task can be cancelled, which can take unprepared users by surprise

Depending on use-case, there are two ways to ensure no surprises:

• usage of onCancelRaiseError, before applying memoization, to ensure that on cancellation an error is triggered and then noticed by the memoization logic
• usage of uncancelable, either before or after applying memoization, to ensure that the memoized task cannot be cancelled

Example:

import scala.concurrent.CancellationException
import scala.concurrent.duration._

val source = Task(1).delayExecution(5.seconds)

// Option 1: trigger error on cancellation
val err = new CancellationException
val cached1 = source.onCancelRaiseError(err).memoizeOnSuccess

// Option 2: make it uninterruptible
val cached2 = source.uncancelable.memoizeOnSuccess

When using onCancelRaiseError like in the example above, the behavior of memoizeOnSuccess is to retry the source on subsequent invocations. Use memoize if that's not the desired behavior.

UNSAFE — this operation allocates a shared, mutable reference, which can break in certain cases referential transparency, even if this operation guarantees idempotency (i.e. referential transparency implies idempotency, but idempotency does not imply referential transparency).

The allocation of a mutable reference is known to be a side effect, thus breaking referential transparency, even if calling this method does not trigger the evaluation of side effects suspended by the source.

Use with care. Sometimes it's easier to just keep a shared, memoized reference to some connection, but keep in mind it might be better to pass such a reference around as a parameter.

Returns a new task that mirrors the source task for normal termination, but that triggers the given error on cancellation.

Normally tasks that are cancelled become non-terminating. Here's an example of a cancelable task:

import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val tenSecs = Task.sleep(10.seconds)
val task1 = tenSecs.start.flatMap { fa =>
  // Triggering pure cancellation, then trying to get its result
  fa.cancel.flatMap(_ => tenSecs)
}

task1.timeout(10.seconds).runToFuture
//=> throws TimeoutException

In general you can expect cancelable tasks to become non-terminating on cancellation.

This onCancelRaiseError operator transforms a task that would yield Task.never on cancellation into one that yields Task.raiseError.

Example:

import java.util.concurrent.CancellationException

val anotherTenSecs = Task.sleep(10.seconds)
  .onCancelRaiseError(new CancellationException)

val task2 = anotherTenSecs.start.flatMap { fa =>
  // Triggering pure cancellation, then trying to get its result
  fa.cancel.flatMap(_ => anotherTenSecs)
}

task2.runToFuture
// => CancellationException

Creates a new task that will handle any matching throwable that this task might emit.

See onErrorRecover for the version that takes a partial function.

Creates a new task that will handle any matching throwable that this task might emit by executing another task.

See onErrorRecoverWith for the version that takes a partial function.

Creates a new task that on error will try to map the error to another value using the provided partial function.

See onErrorHandle for the version that takes a total function.

Creates a new task that will try recovering from an error by matching it with another task using the given partial function.

See onErrorHandleWith for the version that takes a total function.

Creates a new task that in case of error will retry executing the source again and again, until it succeeds.

In case of continuous failure the total number of executions will be maxRetries + 1.

Creates a new task that in case of error will retry executing the source again and again, until it succeeds, or until the given predicate returns false.

In this sample we retry for as long as the exception is a TimeoutException:

import scala.concurrent.TimeoutException

Task("some long call that may timeout").onErrorRestartIf {
  case _: TimeoutException => true
  case _ => false
}

On error restarts the source with a customizable restart loop.

This operation keeps an internal state, with a start value, an internal state that gets evolved and based on which the next step gets decided, e.g. should it restart, maybe with a delay, or should it give up and re-throw the current error.

Example that implements a simple retry policy that retries for a maximum of 10 times before giving up; also introduce a 1 second delay before each retry is executed:

import scala.util.Random
import scala.concurrent.duration._

val task = Task {
  if (Random.nextInt(20) > 10)
    throw new RuntimeException("boo")
  else 78
}

task.onErrorRestartLoop(10) { (err, maxRetries, retry) =>
  if (maxRetries > 0)
    // Next retry please; but do a 1 second delay
    retry(maxRetries - 1).delayExecution(1.second)
  else
    // No retries left, rethrow the error
    Task.raiseError(err)
}

A more complex exponential back-off sample:

import scala.concurrent.duration._

// Keeps the current state, indicating the restart delay and the
// maximum number of retries left
final case class Backoff(maxRetries: Int, delay: FiniteDuration)

// Restarts for a maximum of 10 times, with an initial delay of 1 second,
// a delay that keeps being multiplied by 2
task.onErrorRestartLoop(Backoff(10, 1.second)) { (err, state, retry) =>
  val Backoff(maxRetries, delay) = state
  if (maxRetries > 0)
    retry(Backoff(maxRetries - 1, delay * 2)).delayExecution(delay)
  else
    // No retries left, rethrow the error
    Task.raiseError(err)
}

The given function injects the following parameters:

1. error reference that was thrown 2. the current state, based on which a decision for the retry is made 3. retry: S => Task[B] function that schedules the next retry

Returns a new value that transforms the result of the source, given the recover or map functions, which get executed depending on whether the result is successful or if it ends in error.

This is an optimization on usage of attempt and map, this equivalence being true:

task.redeem(recover, map) <-> task.attempt.map(_.fold(recover, map))

Usage of redeem subsumes onErrorHandle because:

task.redeem(fe, id) <-> task.onErrorHandle(fe)

Returns a new value that transforms the result of the source, given the recover or bind functions, which get executed depending on whether the result is successful or if it ends in error.

This is an optimization on usage of attempt and flatMap, this equivalence being available:

task.redeemWith(recover, bind) <-> task.attempt.flatMap(_.fold(recover, bind))

Usage of redeemWith subsumes onErrorHandleWith because:

task.redeemWith(fe, F.pure) <-> task.onErrorHandleWith(fe)

Usage of redeemWith also subsumes flatMap because:

task.redeemWith(Task.raiseError, fs) <-> task.flatMap(fs)

Triggers the asynchronous execution, with a provided callback that's going to be called at some point in the future with the final result.

Note that without invoking runAsync on a Task, nothing gets evaluated, as a Task has lazy behavior.

import scala.concurrent.duration._
// A Scheduler is needed for executing tasks via `runAsync`
import monix.execution.Scheduler.Implicits.global

// Nothing executes yet
val task: Task[String] =
  for {
    _ <- Task.sleep(3.seconds)
    r <- Task { println("Executing..."); "Hello!" }
  } yield r


// Triggering the task's execution:
val f = task.runAsync {
  case Right(str: String) =>
    println(s"Received: $$str")
  case Left(e) =>
    global.reportFailure(e)
}

// Or in case we change our mind
f.cancel()

Callback

When executing the task via this method, the user is required to supply a side effecting callback with the signature: Either[Throwable, A] => Unit.

This will be used by the implementation to signal completion, signaling either a Right(value) or a Left(error).

Task however uses Callback internally, so you can supply a Callback instance instead and it will be used to avoid unnecessary boxing. It also has handy utilities.

Note that with Callback you can:

• convert from a plain function using Either[Throwable, A] as input via Callback.fromAttempt
• convert from a plain function using Try[A] as input via Callback.fromTry
• wrap a standard Scala Promise via Callback.fromPromise
• pass an empty callback that just reports errors via Callback.empty

Example, equivalent to the above:

import monix.execution.Callback

task.runAsync(new Callback[Throwable, String] {
  def onSuccess(str: String) =
    println(s"Received: $$str")
  def onError(e: Throwable) =
    global.reportFailure(e)
})

Example equivalent with runAsyncAndForget:

task.runAsync(Callback.empty)

Completing a scala.concurrent.Promise:

import scala.concurrent.Promise

val p = Promise[String]()
task.runAsync(Callback.fromPromise(p))

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Triggers the asynchronous execution of the source task in a "fire and forget" fashion.

Starts the execution of the task, but discards any result generated asynchronously and doesn't return any cancelable tokens either. This affords some optimizations — for example the underlying run-loop doesn't need to worry about cancelation. Also the call-site is more clear in intent.

Example:

import monix.execution.Scheduler.Implicits.global

val task = Task(println("Hello!"))

// We don't care about the result, we don't care about the
// cancellation token, we just want this thing to run:
task.runAsyncAndForget

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Triggers the asynchronous execution in a "fire and forget" fashion, like normal runAsyncAndForget, but includes the ability to specify Task.Options that can modify the behavior of the run-loop.

This allows you to specify options such as:

• enabling support for TaskLocal
• disabling auto-cancelable run-loops

See the description of runAsyncOpt for an example of customizing the default Task.Options.

See the description of runAsyncAndForget for an example of running as a "fire and forget".

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Triggers the asynchronous execution, returning a Task[Unit] (aliased to CancelToken[Task] in Cats-Effect) which can cancel the running computation.

This is the more potent version of runAsync, because the returned cancelation token is a Task[Unit] that can be used to back-pressure on the result of the cancellation token, in case the finalizers are specified as asynchronous actions that are expensive to complete.

Example:

import scala.concurrent.duration._

val task = Task("Hello!").bracketCase { str =>
  Task(println(str))
} { (_, exitCode) =>
  // Finalization
  Task(println(s"Finished via exit code: $$exitCode"))
    .delayExecution(3.seconds)
}

In this example we have a task with a registered finalizer (via bracketCase) that takes 3 whole seconds to finish. Via normal runAsync the returned cancelation token has no capability to wait for its completion.

import monix.execution.Callback
import monix.execution.Scheduler.Implicits.global

val cancel = task.runAsyncF(Callback.empty)

// Triggering `cancel` and we can wait for its completion
for (_ <- cancel.runToFuture) {
  // Takes 3 seconds to print
  println("Resources were released!")
}

WARN: back-pressuring on the completion of finalizers is not always a good idea. Avoid it if you can.

Callback

When executing the task via this method, the user is required to supply a side effecting callback with the signature: Either[Throwable, A] => Unit.

This will be used by the implementation to signal completion, signaling either a Right(value) or a Left(error).

Task however uses Callback internally, so you can supply a Callback instance instead and it will be used to avoid unnecessary boxing. It also has handy utilities.

Note that with Callback you can:

• convert from a plain function using Either[Throwable, A] as input via Callback.fromAttempt
• convert from a plain function using Try[A] as input via Callback.fromTry
• wrap a standard Scala Promise via Callback.fromPromise
• pass an empty callback that just reports errors via Callback.empty

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

NOTE: the F suffix comes from F[_], highlighting our usage of CancelToken[F] to return a Task[Unit], instead of a plain and side effectful Cancelable object.

Triggers the asynchronous execution, much like normal runAsync, but includes the ability to specify Task.Options that can modify the behavior of the run-loop.

This allows you to specify options such as:

• enabling support for TaskLocal
• disabling auto-cancelable run-loops

Example:

import monix.execution.Scheduler.Implicits.global

val task =
  for {
    local <- TaskLocal(0)
    _     <- local.write(100)
    _     <- Task.shift
    value <- local.read
  } yield value

// We need to activate support of TaskLocal via:
implicit val opts = Task.defaultOptions.enableLocalContextPropagation

// Actual execution that depends on these custom options:
task.runAsyncOpt {
  case Right(value) =>
    println(s"Received: $$value")
  case Left(e) =>
    global.reportFailure(e)
}

See Task.Options.

Callback

When executing the task via this method, the user is required to supply a side effecting callback with the signature: Either[Throwable, A] => Unit.

This will be used by the implementation to signal completion, signaling either a Right(value) or a Left(error).

Task however uses Callback internally, so you can supply a Callback instance instead and it will be used to avoid unnecessary boxing. It also has handy utilities.

Note that with Callback you can:

• convert from a plain function using Either[Throwable, A] as input via Callback.fromAttempt
• convert from a plain function using Try[A] as input via Callback.fromTry
• wrap a standard Scala Promise via Callback.fromPromise
• pass an empty callback that just reports errors via Callback.empty

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Triggers the asynchronous execution, much like normal runAsyncF, but includes the ability to specify Task.Options that can modify the behavior of the run-loop.

This allows you to specify options such as:

• enabling support for TaskLocal
• disabling auto-cancelable run-loops

See the description of runToFutureOpt for an example.

The returned cancelation token is a Task[Unit] that can be used to back-pressure on the result of the cancellation token, in case the finalizers are specified as asynchronous actions that are expensive to complete.

See the description of runAsyncF for an example.

WARN: back-pressuring on the completion of finalizers is not always a good idea. Avoid it if you can.

Callback

When executing the task via this method, the user is required to supply a side effecting callback with the signature: Either[Throwable, A] => Unit.

This will be used by the implementation to signal completion, signaling either a Right(value) or a Left(error).

Task however uses Callback internally, so you can supply a Callback instance instead and it will be used to avoid unnecessary boxing. It also has handy utilities.

Note that with Callback you can:

• convert from a plain function using Either[Throwable, A] as input via Callback.fromAttempt
• convert from a plain function using Try[A] as input via Callback.fromTry
• wrap a standard Scala Promise via Callback.fromPromise
• pass an empty callback that just reports errors via Callback.empty

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

NOTE: the F suffix comes from F[_], highlighting our usage of CancelToken[F] to return a Task[Unit], instead of a plain and side effectful Cancelable object.

Triggers the asynchronous execution of the source task, but runs it in uncancelable mode.

This is an optimization over plain runAsync or runAsyncF that doesn't give you a cancellation token for cancelling the task. The runtime can thus not worry about keeping state related to cancellation when evaluating it.

import scala.concurrent.duration._
import monix.execution.Scheduler.Implicits.global

val task: Task[String] =
  for {
    _ <- Task.sleep(3.seconds)
    r <- Task { println("Executing..."); "Hello!" }
  } yield r

// Triggering the task's execution, without receiving any
// cancelation tokens
task.runAsyncUncancelable {
  case Right(str) =>
    println(s"Received: $$str")
  case Left(e) =>
    global.reportFailure(e)
}

Callback

When executing the task via this method, the user is required to supply a side effecting callback with the signature: Either[Throwable, A] => Unit.

This will be used by the implementation to signal completion, signaling either a Right(value) or a Left(error).

Task however uses Callback internally, so you can supply a Callback instance instead and it will be used to avoid unnecessary boxing. It also has handy utilities.

Note that with Callback you can:

• convert from a plain function using Either[Throwable, A] as input via Callback.fromAttempt
• convert from a plain function using Try[A] as input via Callback.fromTry
• wrap a standard Scala Promise via Callback.fromPromise
• pass an empty callback that just reports errors via Callback.empty

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Triggers the asynchronous execution in uncancelable mode, like runAsyncUncancelable, but includes the ability to specify Task.Options that can modify the behavior of the run-loop.

This allows you to specify options such as:

• enabling support for TaskLocal
• disabling auto-cancelable run-loops

See the description of runAsyncOpt for an example of customizing the default Task.Options.

This is an optimization over plain runAsyncOpt or runAsyncOptF that doesn't give you a cancellation token for cancelling the task. The runtime can thus not worry about keeping state related to cancellation when evaluating it.

Callback

When executing the task via this method, the user is required to supply a side effecting callback with the signature: Either[Throwable, A] => Unit.

This will be used by the implementation to signal completion, signaling either a Right(value) or a Left(error).

Task however uses Callback internally, so you can supply a Callback instance instead and it will be used to avoid unnecessary boxing. It also has handy utilities.

Note that with Callback you can:

• convert from a plain function using Either[Throwable, A] as input via Callback.fromAttempt
• convert from a plain function using Try[A] as input via Callback.fromTry
• wrap a standard Scala Promise via Callback.fromPromise
• pass an empty callback that just reports errors via Callback.empty

Executes the source until completion, or until the first async boundary, whichever comes first.

This operation is mean to be compliant with cats.effect.Effect.runSyncStep, but without suspending the evaluation in IO.

WARNING: This method is a partial function, throwing exceptions in case errors happen immediately (synchronously).

Usage sample:

import monix.execution.Scheduler.Implicits.global
import scala.util._
import scala.util.control.NonFatal

try Task(42).runSyncStep match {
  case Right(a) => println("Success: " + a)
  case Left(task) =>
    task.runToFuture.onComplete {
      case Success(a) => println("Async success: " + a)
      case Failure(e) => println("Async error: " + e)
    }
} catch {
  case NonFatal(e) =>
    println("Error: " + e)
}

Obviously the purpose of this method is to be used for optimizations.

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

A variant of runSyncStep that takes an implicit Task.Options from the current scope.

This helps in tuning the evaluation model of task.

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Evaluates the source task synchronously and returns the result immediately or blocks the underlying thread until the result is ready.

WARNING: blocking operations are unsafe and incredibly error prone on top of the JVM. It's a good practice to not block any threads and use the asynchronous runAsync methods instead.

In general prefer to use the asynchronous Task.runAsync or Task.runToFuture and to structure your logic around asynchronous actions in a non-blocking way. But in case you're blocking only once, in main, at the "edge of the world" so to speak, then it's OK.

Sample:

import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

Task(42).runSyncUnsafe(3.seconds)

This is equivalent with:

import scala.concurrent.Await

Await.result[Int](Task(42).runToFuture, 3.seconds)

Some implementation details:

• blocking the underlying thread is done by triggering Scala's BlockingContext (scala.concurrent.blocking), just like Scala's Await.result
• the timeout is mandatory, just like when using Scala's Await.result, in order to make the caller aware that the operation is dangerous and that setting a timeout is good practice
• the loop starts in an execution mode that ignores BatchedExecution or AlwaysAsyncExecution, until the first asynchronous boundary. This is because we want to block the underlying thread for the result, in which case preserving fairness by forcing (batched) async boundaries doesn't do us any good, quite the contrary, the underlying thread being stuck until the result is available or until the timeout exception gets triggered.

Not supported on top of JavaScript engines and trying to use it with Scala.js will trigger a compile time error.

For optimizations on top of JavaScript you can use runSyncStep instead.

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Variant of runSyncUnsafe that takes a Task.Options implicitly from the scope in order to tune the evaluation model of the task.

This allows you to specify options such as:

• enabling support for TaskLocal
• disabling auto-cancelable run-loops

See the description of runAsyncOpt for an example of customizing the default Task.Options.

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

Triggers the asynchronous execution, returning a cancelable CancelableFuture that can be awaited for the final result or canceled.

Note that without invoking runAsync on a Task, nothing gets evaluated, as a Task has lazy behavior.

import scala.concurrent.duration._
// A Scheduler is needed for executing tasks via `runAsync`
import monix.execution.Scheduler.Implicits.global

// Nothing executes yet
val task: Task[String] =
  for {
    _ <- Task.sleep(3.seconds)
    r <- Task { println("Executing..."); "Hello!" }
  } yield r


// Triggering the task's execution:
val f = task.runToFuture

// Or in case we change our mind
f.cancel()

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

BAD CODE:

import monix.execution.CancelableFuture
import scala.concurrent.Await

// ANTI-PATTERN 1: Unnecessary side effects
def increment1(sample: Task[Int]): CancelableFuture[Int] = {
  // No reason to trigger `runAsync` for this operation
  sample.runToFuture.map(_ + 1)
}

// ANTI-PATTERN 2: blocking threads makes it worse than (1)
def increment2(sample: Task[Int]): Int = {
  // Blocking threads is totally unnecessary
  val x = Await.result(sample.runToFuture, 5.seconds)
  x + 1
}

// ANTI-PATTERN 3: this is even WORSE than (2)!
def increment3(sample: Task[Int]): Task[Int] = {
  // Triggering side-effects, but misleading users/readers
  // into thinking this function is pure via the return type
  Task.fromFuture(sample.runToFuture.map(_ + 1))
}

Instead prefer the pure versions. Task has its own map, flatMap, onErrorHandleWith or bracketCase, which are really powerful and can allow you to operate on a task in however way you like without escaping Task's context and triggering unwanted side-effects.

Triggers the asynchronous execution, much like normal runToFuture, but includes the ability to specify Options that can modify the behavior of the run-loop.

This is the configurable version of runToFuture. It allows you to specify options such as:

• enabling support for TaskLocal
• disabling auto-cancelable run-loops

See Task.Options. Example:

import monix.execution.Scheduler.Implicits.global

val task =
  for {
    local <- TaskLocal(0)
    _     <- local.write(100)
    _     <- Task.shift
    value <- local.read
  } yield value

// We need to activate support of TaskLocal via:
implicit val opts = Task.defaultOptions.enableLocalContextPropagation
// Actual execution that depends on these custom options:
task.runToFutureOpt

UNSAFE (referential transparency) — this operation can trigger the execution of side effects, which breaks referential transparency and is thus not a pure function.

Normally these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request (in case you're working with a web framework or toolkit that doesn't provide good integration with Monix's Task via Cats-Effect).

Otherwise for modifying or operating on tasks, prefer its pure functions like map and flatMap. In FP code don't use runAsync. Remember that Task is not a 1:1 replacement for Future, Task being a very different abstraction.

PLEASE READ the advice on anti-patterns at runToFuture.

Start execution of the source suspended in the Task context.

This can be used for non-deterministic / concurrent execution. The following code is more or less equivalent with Task.parMap2 (minus the behavior on error handling and cancellation):

def par2[A, B](ta: Task[A], tb: Task[B]): Task[(A, B)] =
  for {
    fa <- ta.start
    fb <- tb.start
     a <- fa.join
     b <- fb.join
  } yield (a, b)

Note in such a case usage of parMap2 (and parMap3, etc.) is still recommended because of behavior on error and cancellation — consider that in the example above, if the first task finishes in error, the second task doesn't get cancelled.

This operation forces an asynchronous boundary before execution

Start asynchronous execution of the source suspended in the Task context, running it in the background and discarding the result.

Similar to start after mapping result to Unit. Below law holds:

task.startAndForget <-> task.start.map(_ => ())

Times the Task execution and returns its duration and the computed value.

Basic usage example:

for {
  r <- Task(1 + 1).timed
  (duration, value) = r
  _ <- Task(println("executed in " + duration.toMillis + " ms"))
} yield value

Returns a Task that mirrors the source Task but that triggers a TimeoutException in case the given duration passes without the task emitting any item.

Useful when timeout is variable, e.g. when task is running in a loop with deadline semantics.

Example:

import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val deadline = 10.seconds.fromNow

val singleCallTimeout = 2.seconds

// re-evaluate deadline time on every request
val actualTimeout = Task(singleCallTimeout.min(deadline.timeLeft))

// expensive remote call
def call(): Unit = ()

val remoteCall = Task(call())
  .timeoutToL(actualTimeout, Task.unit)
  .onErrorRestart(100)
  .timeout(deadline.time)

Returns a Task that mirrors the source Task but switches to the given backup Task in case the given duration passes without the source emitting any item.

Useful when timeout is variable, e.g. when task is running in a loop with deadline semantics.

Example:

import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val deadline = 10.seconds.fromNow

val singleCallTimeout = 2.seconds

// re-evaluate deadline time on every request
val actualTimeout = Task(singleCallTimeout.min(deadline.timeLeft))

// expensive remote call
def call(): Unit = ()

val remoteCall = Task(call())
  .timeoutL(actualTimeout)
  .onErrorRestart(100)
  .timeout(deadline.time)

Note that this method respects the timeout task evaluation duration, e.g. if it took 3 seconds to evaluate after to a value of 5 seconds, then this task will timeout in exactly 5 seconds from the moment computation started, which means in 2 seconds after the timeout task has been evaluated.

Returns a Task that mirrors the source Task but that triggers a specified Exception in case the given duration passes without the task emitting any item.

Generic conversion of Task to any data type for which there's a TaskLift implementation available.

Supported data types:

• cats.effect.IO
• any data type implementing cats.effect.Concurrent
• any data type implementing cats.effect.Async
• any data type implementing cats.effect.LiftIO
• monix.reactive.Observable
• monix.tail.Iterant

This conversion guarantees:

• referential transparency
• similar runtime characteristics (e.g. if the source doesn't block threads on evaluation, then the result shouldn't block threads either)
• interruptibility, if the target data type is cancelable

Sample:

import cats.effect.IO
import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

Task(1 + 1)
  .delayExecution(5.seconds)
  .to[IO]

Converts the source Task to any data type that implements Async.

Example:

import cats.effect.IO
import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

Task.eval(println("Hello!"))
  .delayExecution(5.seconds)
  .toAsync[IO]

An Effect[Task] instance is needed in scope, which might need a Scheduler to be available. Such a requirement is needed because the Task has to be evaluated in order to be converted.

NOTE: the resulting instance will NOT be cancelable, as in Task's cancelation token doesn't get carried over. This is implicit in the usage of cats.effect.Async type class. In the example above what this means is that the task will still print "Hello!" after 5 seconds, even if the resulting task gets cancelled.

Converts the source Task to any data type that implements Concurrent.

Example:

import cats.effect.IO
import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

Task.eval(println("Hello!"))
  .delayExecution(5.seconds)
  .to[IO]

A ConcurrentEffect instance for Task is also needed in scope, which might need a Scheduler to be available. Such a requirement is needed because the Task has to be evaluated in order to be converted.

NOTE: the resulting value is cancelable, via usage of cats.effect.Concurrent.

Converts a Task to an org.reactivestreams.Publisher that emits a single item on success, or just the error on failure.

See reactive-streams.org for the Reactive Streams specification.

Returns a string representation of this task meant for debugging purposes only.

Makes the source Task uninterruptible such that a cancel signal (e.g. Fiber.cancel) has no effect.

import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val uncancelable = Task
  .eval(println("Hello!"))
  .delayExecution(10.seconds)
  .uncancelable
  .runToFuture

// No longer works
uncancelable.cancel()

// After 10 seconds
// => Hello!

DEPRECATED — subsumed by startAndForget.

Renamed to startAndForget to be consistent with start which also enforces an asynchronous boundary