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.
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.eval or
Task.apply:
val hello = Task.eval("Hello ") val world = Task("World!")
Nothing gets executed yet, as Task is lazy, nothing executes
until you trigger .runAsync on it.
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 val f: CancelableFuture[Unit] = sayHello.run() //=> 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
The fact that Task is lazy whereas Future is not
has real consequences. For example with Task you can do this:
def retryOnFailure[A](times: Int, source: Task[A]): Task[A] = source.recoverWith { err => // No more retries left? Re-throw error: if (times <= 0) Task.raise(err) else { // Recursive call, yes we can! retryOnFailure(times - 1, source) // Adding 500 ms delay for good measure .delayExecution(500) } }
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 can also
do memoization of course:
task.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:
task.memoizeOnSuccess
This keeps repeating the computation for as long as the result is a failure and caches it only on success. Yes we can!
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]] = ??? // Split our list in chunks of 30 items per chunk, // this being the maximum parallelism allowed val chunks = list.sliding(30, 30) // 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 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.async (alias
Task.create):
import scala.concurrent.duration._ val delayedHello = Task.async { (scheduler, callback) => val task = scheduler.scheduleOnce(1.second) { println("Delayed Hello!") // Signaling successful completion callback(Success(())) } Cancelable { () => { 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:
// Triggering execution val f: CancelableFuture[Unit] = delayedHello.run() // If we change our mind before the timespan has passed: f.cancel()
Also, given an Task task, we can specify actions that need to be
triggered in case of cancellation:
val task = Task.eval(println("Hello!")).executeWithFork task.doOnCancel(Task.eval { println("A cancellation attempt was made!") } val f: CancelableFuture[Unit] = task.run() // Note that in this case cancelling the resulting Future // will not stop the actual execution, since it doesn't know // how, but it will trigger our on-cancel callback: f.cancel() //=> A cancellation attempt was made!
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.
- Alphabetic
- By Inheritance
- Task
- Serializable
- Serializable
- AnyRef
- Any
- Hide All
- Show All
- Public
- All
Value Members
-
final
def
!=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
final
def
##(): Int
- Definition Classes
- AnyRef → Any
-
final
def
==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
final
def
asInstanceOf[T0]: T0
- Definition Classes
- Any
-
final
def
asyncBoundary(s: Scheduler): Task[A]
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).
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
ioscheduler, including themaptransformation that follows afterexecuteOn. 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
sourcewill happen on theioscheduler, however theasyncBoundarycall will make all subsequent operations to happen on the specifiedglobalscheduler.- s
is the scheduler triggering the asynchronous boundary
-
final
def
asyncBoundary: Task[A]
Introduces an asynchronous boundary at the current stage in the asynchronous processing pipeline.
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
ioscheduler, including themaptransformation that follows afterexecuteOn. 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
sourcewill happen on theioscheduler, however theasyncBoundarycall will make all subsequent operations to happen on the default scheduler. -
final
def
attempt: Task[Either[Throwable, A]]
Creates a new Task that will expose any triggered error from the source.
-
final
def
cancel: Task[Unit]
Signals cancellation of the source.
Signals cancellation of the source.
Returns a new task that will complete when the cancellation is sent (but not when it is observed).
Compared with CancelableFuture.cancel() this action is pure.
Example:
Task.racePair(ta, tb).flatMap { case Left((a, tb)) => tb.cancel.map(_ => a) case Right((ta, b)) => ta.cancel.map(_ => b) } -
def
clone(): AnyRef
- Attributes
- protected[java.lang]
- Definition Classes
- AnyRef
- Annotations
- @native() @throws( ... )
-
final
def
coeval(implicit s: Scheduler): Coeval[Either[CancelableFuture[A], A]]
Transforms a Task into a Coeval that tries to execute the source synchronously, returning either
Right(value)in case a value is available immediately, orLeft(future)in case we have an asynchronous boundary or an error. -
final
def
delayExecution(timespan: FiniteDuration): Task[A]
Returns a task that waits for the specified
timespanbefore executing and mirroring the result of the source.Returns a task that waits for the specified
timespanbefore executing and mirroring the result of the source.- See also
delayExecutionWith for delaying the execution of the source with a customizable trigger.
-
final
def
delayExecutionWith(trigger: Task[Any]): Task[A]
Returns a task that waits for the specified
triggerto succeed before mirroring the result of the source.Returns a task that waits for the specified
triggerto succeed before mirroring the result of the source.If the
triggerends in error, then the resulting task will also end in error.As an example, these are equivalent (in the observed effects and result, not necessarily in implementation):
val ta = source.delayExecution(10.seconds) val tb = source.delayExecutionWith(Task.unit.delayExecution(10.seconds))
- See also
delayExecution for delaying the execution of the source with a simple timespan
-
final
def
delayResult(timespan: FiniteDuration): Task[A]
Returns a task that executes the source immediately on
runAsync, but before emitting theonSuccessresult for the specified duration.Returns a task that executes the source immediately on
runAsync, but before emitting theonSuccessresult for the specified duration.Note that if an error happens, then it is streamed immediately with no delay.
- See also
delayResultBySelector for applying different delay strategies depending on the signaled result.
-
final
def
delayResultBySelector[B](selector: (A) ⇒ Task[B]): Task[A]
Returns a task that executes the source immediately on
runAsync, but with the result delayed by the specifiedselector.Returns a task that executes the source immediately on
runAsync, but with the result delayed by the specifiedselector.The
selectorgenerates anotherTaskwhose execution will delay the signaling of the result generated by the source. Compared with delayResult this gives you an opportunity to apply different delay strategies depending on the signaled result.As an example, these are equivalent (in the observed effects and result, not necessarily in implementation):
val t1 = source.delayResult(10.seconds) val t2 = source.delayResultBySelector(_ => Task.unit.delayExecution(10.seconds))
Note that if an error happens, then it is streamed immediately with no delay.
- See also
delayResult for delaying with a simple timeout
-
final
def
dematerialize[B](implicit ev: <:<[A, Try[B]]): Task[B]
Dematerializes the source's result from a
Try. -
final
def
doOnCancel(callback: Task[Unit]): Task[A]
Returns a new
Taskthat will mirror the source, but that will execute the givencallbackif the task gets canceled before completion.Returns a new
Taskthat will mirror the source, but that will execute the givencallbackif the task gets canceled before completion.This only works for premature cancellation. See doOnFinish for triggering callbacks when the source finishes.
- callback
is the callback to execute if the task gets canceled prematurely
-
final
def
doOnFinish(f: (Option[Throwable]) ⇒ Task[Unit]): Task[A]
Returns a new
Taskin whichfis scheduled to be run on completion.Returns a new
Taskin whichfis scheduled to be run on completion. This would typically be used to release any resources acquired by thisTask.The returned
Taskcompletes when both the source and the task returned byfcomplete.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.
-
final
def
eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
-
def
equals(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
final
def
executeOn(s: Scheduler, forceAsync: Boolean = true): Task[A]
Overrides the default Scheduler, possibly forcing an asynchronous boundary before execution (if
forceAsyncis set totrue, the default).Overrides the default Scheduler, possibly forcing an asynchronous boundary before execution (if
forceAsyncis set totrue, the default).When a
Taskis executed with .runAsync, it needs aScheduler, which is going to be injected in all asynchronous tasks processed within theflatMapchain, aSchedulerthat is used to manage asynchronous boundaries and delayed execution.This scheduler passed in
runAsyncis said to be the "default" andexecuteOnoverrides 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
thunkimmediately (on the current thread and call stack).By calling
executeOn(io), we are ensuring that the usedScheduler(injected in async tasks by means of Task.Context) will beio, aSchedulerthat we intend to use for blocking I/O actions. And we are also forcing an asynchronous boundary right before execution, by passing theforceAsyncparameter astrue(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
ioscheduler, 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 givenScheduler, that depending of the implementation of theTask. 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
executeOnwe'll indeed get a logical fork on theioscheduler: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.executeOn(io1).executeOn(io2)
In this example the implementation of
taskwill receive the reference toioand will use it on evaluation, while the second invocation ofexecuteOnwill create an unnecessary async boundary (ifforceAsync = true) or be basically a costly no-op. This might be confusing but consider the equivalence to these functions:import scala.concurrent.ExecutionContext val io1 = Scheduler.io() val io2 = 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(io) def sayHello3(ec: ExecutionContext): Unit = // Overriding the default no longer has the desired effect // because sayHello2 is ignoring it! sayHello2(io2)
- s
is the Scheduler to use for overriding the default scheduler and for forcing an asynchronous boundary if
forceAsyncistrue- forceAsync
indicates whether an asynchronous boundary should be forced right before the evaluation of the
Task, managed by the providedScheduler- returns
a new
Taskthat mirrors the source on evaluation, but that uses the provided scheduler for overriding the default and possibly force an extra asynchronous boundary on execution
-
final
def
executeWithFork: Task[A]
Mirrors the given source
Task, but upon execution ensure that evaluation forks into a separate (logical) thread.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 .runAsync.
This operation is equivalent with:
Task.shift.flatMap(_ => task) // ... or ... import cats.syntax.all._ Task.shift.followedBy(task)
The Scheduler used for scheduling the async boundary will be the default, meaning the one used to start the run-loop in
runAsync. -
final
def
executeWithModel(em: ExecutionModel): Task[A]
Returns a new task that will execute the source with a different ExecutionModel.
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.executeWithModel(AlwaysAsyncExecution)- em
is the ExecutionModel with which the source will get evaluated on
runAsync
-
final
def
executeWithOptions(f: (Options) ⇒ Options): Task[A]
Returns a new task that will execute the source with a different set of Options.
Returns a new task that will execute the source with a different set of Options.
This allows fine-tuning the default options. Example:
task.executeWithOptions(_.enableAutoCancelableRunLoops)
- f
is a function that takes the source's current set of options and returns a modified set of options that will be used to execute the source upon
runAsync
-
final
def
failed: Task[Throwable]
Returns a failed projection of this task.
Returns a failed projection of this task.
The failed projection is a
Taskholding a value of typeThrowable, emitting the error yielded by the source, in case the source fails, otherwise if the source succeeds the result will fail with aNoSuchElementException. -
def
finalize(): Unit
- Attributes
- protected[java.lang]
- Definition Classes
- AnyRef
- Annotations
- @throws( classOf[java.lang.Throwable] )
-
final
def
flatMap[B](f: (A) ⇒ Task[B]): Task[B]
Creates a new Task by applying a function to the successful result of the source Task, and returns a task equivalent to the result of the function.
-
final
def
flatten[B](implicit ev: <:<[A, Task[B]]): Task[B]
Given a source Task that emits another Task, this function flattens the result, returning a Task equivalent to the emitted Task by the source.
-
final
def
foreach(f: (A) ⇒ Unit)(implicit s: Scheduler): CancelableFuture[Unit]
Triggers the evaluation of the source, executing the given function for the generated element.
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.
-
final
def
foreachL(f: (A) ⇒ Unit): Task[Unit]
Returns a new task that upon evaluation will execute the given function for the generated element, transforming the source into a
Task[Unit].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.
-
final
def
getClass(): Class[_]
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
-
def
hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
-
final
def
isInstanceOf[T0]: Boolean
- Definition Classes
- Any
-
final
def
map[B](f: (A) ⇒ B): Task[B]
Returns a new
Taskthat applies the mapping function to the element emitted by the source.Returns a new
Taskthat 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:
scala fa.map(f) <-> fa.flatMap(x => Task.pure(f(x))) -
final
def
materialize: Task[Try[A]]
Creates a new Task that will expose any triggered error from the source.
-
final
def
memoize: Task[A]
Memoizes (caches) the result of the source task and reuses it on subsequent invocations of
runAsync.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.
- See also
memoizeOnSuccess for a version that only caches successful results
-
final
def
memoizeOnSuccess: Task[A]
Memoizes (cache) the successful result of the source task and reuses it on subsequent invocations of
runAsync.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.
- See also
memoize for a version that caches both successful results and failures
-
final
def
ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
-
final
def
notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
-
final
def
notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
-
final
def
onErrorFallbackTo[B >: A](that: Task[B]): Task[B]
Creates a new task that in case of error will fallback to the given backup task.
-
final
def
onErrorHandle[U >: A](f: (Throwable) ⇒ U): Task[U]
Creates a new task that will handle any matching throwable that this task might emit.
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.
-
final
def
onErrorHandleWith[B >: A](f: (Throwable) ⇒ Task[B]): Task[B]
Creates a new task that will handle any matching throwable that this task might emit by executing another task.
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.
-
final
def
onErrorRecover[U >: A](pf: PartialFunction[Throwable, U]): Task[U]
Creates a new task that on error will try to map the error to another value using the provided 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.
-
final
def
onErrorRecoverWith[B >: A](pf: PartialFunction[Throwable, Task[B]]): Task[B]
Creates a new task that will try recovering from an error by matching it with another task using the given partial 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.
-
final
def
onErrorRestart(maxRetries: Long): Task[A]
Creates a new task that in case of error will retry executing the source again and again, until it succeeds.
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. -
final
def
onErrorRestartIf(p: (Throwable) ⇒ Boolean): Task[A]
Creates a new task that in case of error will retry executing the source again and again, until it succeeds.
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. -
final
def
restartUntil(p: (A) ⇒ Boolean): Task[A]
Given a predicate function, keep retrying the task until the function returns true.
-
def
runAsync(cb: Callback[A])(implicit s: Scheduler): Cancelable
Triggers the asynchronous execution.
Triggers the asynchronous execution.
Without invoking
runAsyncon aTask, nothing gets evaluated, as aTaskhas lazy behavior.- cb
is a callback that will be invoked upon completion
- s
is an injected Scheduler that gets used whenever asynchronous boundaries are needed when evaluating the task
- returns
a Cancelable that can be used to cancel a running task
-
def
runAsync(implicit s: Scheduler): CancelableFuture[A]
Triggers the asynchronous execution.
Triggers the asynchronous execution.
Without invoking
runAsyncon aTask, nothing gets evaluated, as aTaskhas lazy behavior.- s
is an injected Scheduler that gets used whenever asynchronous boundaries are needed when evaluating the task
- returns
a CancelableFuture that can be used to extract the result or to cancel a running task.
-
def
runAsyncOpt(cb: Callback[A])(implicit s: Scheduler, opts: Options): Cancelable
Triggers the asynchronous execution, much like normal
runAsync, but includes the ability to specify Options that can modify the behavior of the run-loop.Triggers the asynchronous execution, much like normal
runAsync, but includes the ability to specify Options that can modify the behavior of the run-loop.- cb
is a callback that will be invoked upon completion
- s
is an injected Scheduler that gets used whenever asynchronous boundaries are needed when evaluating the task
- opts
a set of Options that determine the behavior of Task's run-loop.
- returns
a Cancelable that can be used to cancel a running task
-
def
runAsyncOpt(implicit s: Scheduler, opts: Options): CancelableFuture[A]
Triggers the asynchronous execution, much like normal
runAsync, but includes the ability to specify Options that can modify the behavior of the run-loop.Triggers the asynchronous execution, much like normal
runAsync, but includes the ability to specify Options that can modify the behavior of the run-loop.- s
is an injected Scheduler that gets used whenever asynchronous boundaries are needed when evaluating the task
- opts
a set of Options that determine the behavior of Task's run-loop.
- returns
a Cancelable that can be used to cancel a running task
-
final
def
runOnComplete(f: (Try[A]) ⇒ Unit)(implicit s: Scheduler): Cancelable
Similar to Scala's
Future#onComplete, this method triggers the evaluation of aTaskand invokes the given callback whenever the result is available.Similar to Scala's
Future#onComplete, this method triggers the evaluation of aTaskand invokes the given callback whenever the result is available.- f
is a callback that will be invoked upon completion
- s
is an injected Scheduler that gets used whenever asynchronous boundaries are needed when evaluating the task
- returns
a Cancelable that can be used to cancel a running task
-
final
def
runSyncMaybe(implicit s: Scheduler): Either[CancelableFuture[A], A]
Tries to execute the source synchronously.
Tries to execute the source synchronously.
As an alternative to
runAsync, this method tries to execute the source task immediately on the current thread and call-stack.WARNING: This method is a partial function, throwing exceptions in case errors happen immediately (synchronously).
Usage sample:
try task.runSyncMaybe match { case Right(a) => println("Success: " + a) case Left(future) => future.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.
- returns
Right(result)in case a result was processed, orLeft(future)in case an asynchronous boundary was hit and further async execution is needed or in case of failure
-
final
def
start: Task[Task[A]]
Start execution of the source suspended in the
Taskcontext.Start execution of the source suspended in the
Taskcontext.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 b <- fb } 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.
IMPORTANT — this operation does not fork, it does not introduce an asynchronous boundary, so in case the evaluation of a task is immediate, a fork might be needed to make evaluation happen on another thread or stack frame (if it doesn't do that already):
Task.fork(myTask).start
Inspired by FS2, with the difference that this method does not fork automatically, being consistent with Monix's default behavior.
-
final
def
synchronized[T0](arg0: ⇒ T0): T0
- Definition Classes
- AnyRef
-
final
def
timeout(after: FiniteDuration): Task[A]
Returns a Task that mirrors the source Task but that triggers a
TimeoutExceptionin case the given duration passes without the task emitting any item. -
final
def
timeoutTo[B >: A](after: FiniteDuration, backup: Task[B]): Task[B]
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.
-
final
def
toIO(implicit s: Scheduler): IO[A]
Converts the source
Taskto acats.effect.IOvalue. -
final
def
toReactivePublisher(implicit s: Scheduler): Publisher[A]
Converts a Task to an
org.reactivestreams.Publisherthat emits a single item on success, or just the error on failure.Converts a Task to an
org.reactivestreams.Publisherthat emits a single item on success, or just the error on failure.See reactive-streams.org for the Reactive Streams specification.
-
def
toString(): String
Returns a string representation of this task meant for debugging purposes only.
Returns a string representation of this task meant for debugging purposes only.
- Definition Classes
- Task → AnyRef → Any
-
final
def
transform[R](fa: (A) ⇒ R, fe: (Throwable) ⇒ R): Task[R]
Creates a new
Taskby applying the 'fa' function to the successful result of this future, or the 'fe' function to the potential errors that might happen.Creates a new
Taskby applying the 'fa' function to the successful result of this future, or the 'fe' function to the potential errors that might happen.This function is similar with map, except that it can also transform errors and not just successful results.
- fa
function that transforms a successful result of the receiver
- fe
function that transforms an error of the receiver
-
final
def
transformWith[R](fa: (A) ⇒ Task[R], fe: (Throwable) ⇒ Task[R]): Task[R]
Creates a new
Taskby applying the 'fa' function to the successful result of this future, or the 'fe' function to the potential errors that might happen.Creates a new
Taskby applying the 'fa' function to the successful result of this future, or the 'fe' function to the potential errors that might happen.This function is similar with flatMap, except that it can also transform errors and not just successful results.
- fa
function that transforms a successful result of the receiver
- fe
function that transforms an error of the receiver
-
final
def
uncancelable: Task[A]
Makes the source
Taskuninterruptible such that a cancel signal has no effect until it finishes. -
final
def
wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
final
def
wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
final
def
wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @throws( ... )
-
final
def
zip[B](that: Task[B]): Task[(A, B)]
Zips the values of
thisandthattask, and creates a new task that will emit the tuple of their results. -
final
def
zipMap[B, C](that: Task[B])(f: (A, B) ⇒ C): Task[C]
Zips the values of
thisandthatand applies the given mapping function on their results.