final class TracingSchedulerService extends Base with SchedulerService
The TracingScheduler
is a Scheduler
implementation that wraps another
SchedulerService
reference, with the purpose of propagating the
Local.Context on async
execution.
- Self Type
- TracingSchedulerService
- Source
- TracingSchedulerService.scala
- Alphabetic
- By Inheritance
- TracingSchedulerService
- SchedulerService
- Base
- BatchingScheduler
- Scheduler
- Executor
- UncaughtExceptionReporter
- Serializable
- ExecutionContext
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Instance Constructors
- new TracingSchedulerService(underlying: SchedulerService)
- underlying
the SchedulerService in charge of the actual execution and scheduling
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
- def awaitTermination(timeout: Long, unit: TimeUnit, awaitOn: ExecutionContext): Future[Boolean]
Returns a
Future
that will be complete when all tasks have completed execution after a shutdown request, or thetimeout
occurs, or the thread awaiting the shutdown is interrupted, whichever happens first.Returns a
Future
that will be complete when all tasks have completed execution after a shutdown request, or thetimeout
occurs, or the thread awaiting the shutdown is interrupted, whichever happens first.NOTE that this method does not block the current thread, unlike the similarly named method in Java's
ExecutionService
. This is because Monix has a strict non-blocking policy, due to the fact that other platforms like Javascript cannot block threads.Because of the non-blocking requirement, this method returns a
Future
result. And on top of the JVM in order to block on such a result, you can just use Scala'sAwait.result
:import scala.concurrent.Await import scala.concurrent.duration._ import monix.execution.Scheduler.global val wasTerminated = Await.result(service.awaitTermination(30.seconds, global), Duration.Inf)
Given the asynchronous execution requirement, the
awaitOn
parameter is a Scheduler that's going to be used for terminating this service and completing ourFuture
. Obviously we cannot reuse this service for awaiting on termination, but Monix's Scheduler.global can always be used for this.- timeout
the maximum time to wait
- unit
the time unit of the timeout argument
- awaitOn
the
ExecutionContext
used for awaiting the shutdown- returns
a
Future
signalingtrue
if this scheduler terminated orfalse
if the timeout elapsed before termination
- Definition Classes
- TracingSchedulerService → SchedulerService
- final def awaitTermination(timeout: FiniteDuration)(implicit permit: CanBlock): Boolean
A blocking version of awaitTermination(timeout:Long* that blocks the current thread.
A blocking version of awaitTermination(timeout:Long* that blocks the current thread.
Due to requiring a CanBlock permit, calls to this function want compile on top of JavaScript, since blocking operations are not supported for JS engines.
- Definition Classes
- SchedulerService
- final def awaitTermination(timeout: FiniteDuration, awaitOn: ExecutionContext): Future[Boolean]
Overload of awaitTermination(timeout:Long*.
Overload of awaitTermination(timeout:Long*.
- Definition Classes
- SchedulerService
- def clockMonotonic(unit: TimeUnit): Long
Returns a monotonic clock measurement, if supported by the underlying platform.
Returns a monotonic clock measurement, if supported by the underlying platform.
This is the pure equivalent of Java's
System.nanoTime
, or ofCLOCK_MONOTONIC
from Linux'sclock_gettime()
.timer.clockMonotonic(NANOSECONDS)
The returned value can have nanoseconds resolution and represents the number of time units elapsed since some fixed but arbitrary origin time. Usually this is the Unix epoch, but that's not a guarantee, as due to the limits of
Long
this will overflow in the future (263 is about 292 years in nanoseconds) and the implementation reserves the right to change the origin.The return value should not be considered related to wall-clock time, the primary use-case being to take time measurements and compute differences between such values, for example in order to measure the time it took to execute a task.
As a matter of implementation detail, Monix's
Scheduler
implementations useSystem.nanoTime
and the JVM will useCLOCK_MONOTONIC
when available, instead ofCLOCK_REALTIME
(seeclock_gettime()
on Linux) and it is up to the underlying platform to implement it correctly.And be warned, there are platforms that don't have a correct implementation of
CLOCK_MONOTONIC
. For example at the moment of writing there is no standard way for such a clock on top of JavaScript and the situation isn't so clear cut for the JVM either, see:- bug report
- concurrency-interest discussion on the X86 tsc register
The JVM tries to do the right thing and at worst the resolution and behavior will be that of
System.currentTimeMillis
.The recommendation is to use this monotonic clock when doing measurements of execution time, or if you value monotonically increasing values more than a correspondence to wall-time, or otherwise prefer clockRealTime.
- Definition Classes
- Base → Scheduler
- final def clockRealTime(unit: TimeUnit): Long
Returns the current time, as a Unix timestamp (number of time units since the Unix epoch).
Returns the current time, as a Unix timestamp (number of time units since the Unix epoch).
This is the equivalent of Java's
System.currentTimeMillis
, or ofCLOCK_REALTIME
from Linux'sclock_gettime()
.The provided
TimeUnit
determines the time unit of the output, its precision, but not necessarily its resolution, which is implementation dependent. For example this will return the number of milliseconds since the epoch:import scala.concurrent.duration.MILLISECONDS scheduler.clockRealTime(MILLISECONDS)
N.B. the resolution is limited by the underlying implementation and by the underlying CPU and OS. If the implementation uses
System.currentTimeMillis
, then it can't have a better resolution than 1 millisecond, plus depending on underlying runtime (e.g. Node.js) it might return multiples of 10 milliseconds or more.See clockMonotonic, for fetching a monotonic value that may be better suited for doing time measurements.
- Definition Classes
- Base → Scheduler
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @native() @HotSpotIntrinsicCandidate()
- final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- final def execute(runnable: Runnable): Unit
Schedules the given
command
for execution at some time in the future.Schedules the given
command
for execution at some time in the future.The command may execute in a new thread, in a pooled thread, in the calling thread, basically at the discretion of the Scheduler implementation.
- Definition Classes
- BatchingScheduler → Scheduler → Executor → ExecutionContext
- final def executeAsync(r: Runnable): Unit
- Definition Classes
- Base → BatchingScheduler
- final def executeAsyncBatch(cb: TrampolinedRunnable): Unit
Schedules the given callback for asynchronous execution in the thread-pool, but also indicates the start of a thread-local trampoline in case the scheduler is a BatchingScheduler.
Schedules the given callback for asynchronous execution in the thread-pool, but also indicates the start of a thread-local trampoline in case the scheduler is a BatchingScheduler.
This utility is provided as an optimization. If you don't understand what this does, then don't worry about it.
- cb
the callback to execute asynchronously
- Definition Classes
- Scheduler
- final def executeTrampolined(cb: TrampolinedRunnable): Unit
Schedules the given callback for immediate execution as a TrampolinedRunnable.
Schedules the given callback for immediate execution as a TrampolinedRunnable. Depending on the execution context, it might get executed on the current thread by using an internal trampoline, so it is still safe from stack-overflow exceptions.
- cb
the callback to execute asynchronously
- Definition Classes
- Scheduler
- final def executionModel: ExecutionModel
The ExecutionModel is a specification of how run-loops and producers should behave in regards to executing tasks either synchronously or asynchronously.
The ExecutionModel is a specification of how run-loops and producers should behave in regards to executing tasks either synchronously or asynchronously.
- Definition Classes
- Base → Scheduler
- final val features: Features
Exposes a set of flags that describes the Scheduler's features.
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- def isShutdown: Boolean
Returns
true
if this scheduler has been shut down.Returns
true
if this scheduler has been shut down.- Definition Classes
- TracingSchedulerService → SchedulerService
- def isTerminated: Boolean
Returns
true
if all tasks have completed following shut down.Returns
true
if all tasks have completed following shut down. Note thatisTerminated
is nevertrue
unless shutdown was called first.- Definition Classes
- TracingSchedulerService → SchedulerService
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def reportFailure(t: Throwable): Unit
Reports that an asynchronous computation failed.
Reports that an asynchronous computation failed.
- Definition Classes
- Base → Scheduler → UncaughtExceptionReporter → ExecutionContext
- final def scheduleAtFixedRate(initialDelay: Long, period: Long, unit: TimeUnit, r: Runnable): Cancelable
Schedules a periodic task that becomes enabled first after the given initial delay, and subsequently with the given period.
Schedules a periodic task that becomes enabled first after the given initial delay, and subsequently with the given period. Executions will commence after
initialDelay
theninitialDelay + period
, theninitialDelay + 2 * period
and so on.If any execution of the task encounters an exception, subsequent executions are suppressed. Otherwise, the task will only terminate via cancellation or termination of the scheduler. If any execution of this task takes longer than its period, then subsequent executions may start late, but will not concurrently execute.
For example the following schedules a message to be printed to standard output approximately every 10 seconds with an initial delay of 5 seconds:
val task = scheduler.scheduleAtFixedRate(5, 10, TimeUnit.SECONDS, new Runnable { def run() = print("Repeated message") }) // later if you change your mind ... task.cancel()
- initialDelay
is the time to wait until the first execution happens
- period
is the time to wait between 2 successive executions of the task
- unit
is the time unit used for the
initialDelay
and theperiod
parameters- r
is the callback to be executed
- returns
a cancelable that can be used to cancel the execution of this repeated task at any time.
- Definition Classes
- Base → Scheduler
- final def scheduleAtFixedRate(initialDelay: FiniteDuration, period: FiniteDuration)(action: => Unit): Cancelable
Schedules a periodic task that becomes enabled first after the given initial delay, and subsequently with the given period.
Schedules a periodic task that becomes enabled first after the given initial delay, and subsequently with the given period. Executions will commence after
initialDelay
theninitialDelay + period
, theninitialDelay + 2 * period
and so on.If any execution of the task encounters an exception, subsequent executions are suppressed. Otherwise, the task will only terminate via cancellation or termination of the scheduler. If any execution of this task takes longer than its period, then subsequent executions may start late, but will not concurrently execute.
For example the following schedules a message to be printed to standard output approximately every 10 seconds with an initial delay of 5 seconds:
val task = scheduler.scheduleAtFixedRate(5.seconds, 10.seconds) { print("Repeated message") } // later if you change your mind ... task.cancel()
- initialDelay
is the time to wait until the first execution happens
- period
is the time to wait between 2 successive executions of the task
- action
is the callback to be executed
- returns
a cancelable that can be used to cancel the execution of this repeated task at any time.
- Definition Classes
- Scheduler
- final def scheduleOnce(initialDelay: Long, unit: TimeUnit, r: Runnable): Cancelable
Schedules a task to run in the future, after
initialDelay
.Schedules a task to run in the future, after
initialDelay
.For example the following schedules a message to be printed to standard output after 5 minutes:
val task = scheduler.scheduleOnce(5, TimeUnit.MINUTES, new Runnable { def run() = print("Hello, world!") }) // later if you change your mind ... task.cancel()
- initialDelay
is the time to wait until the execution happens
- unit
is the time unit used for
initialDelay
- r
is the callback to be executed
- returns
a
Cancelable
that can be used to cancel the created task before execution.
- Definition Classes
- Base → Scheduler
- final def scheduleOnce(initialDelay: FiniteDuration)(action: => Unit): Cancelable
Schedules a task to run in the future, after
initialDelay
.Schedules a task to run in the future, after
initialDelay
.For example the following schedules a message to be printed to standard output after 5 minutes:
val task = scheduler.scheduleOnce(5.minutes) { print("Hello, world!") } // later, if you change your mind ... task.cancel()
- initialDelay
is the time to wait until the execution happens
- action
is the callback to be executed
- returns
a
Cancelable
that can be used to cancel the created task before execution.
- Definition Classes
- Scheduler
- final def scheduleWithFixedDelay(initialDelay: Long, delay: Long, unit: TimeUnit, r: Runnable): Cancelable
Schedules for execution a periodic task that is first executed after the given initial delay and subsequently with the given delay between the termination of one execution and the commencement of the next.
Schedules for execution a periodic task that is first executed after the given initial delay and subsequently with the given delay between the termination of one execution and the commencement of the next.
For example the following schedules a message to be printed to standard output every 10 seconds with an initial delay of 5 seconds:
val task = s.scheduleWithFixedDelay(5, 10, TimeUnit.SECONDS, new Runnable { def run() = print("Repeated message") }) // later if you change your mind ... task.cancel()
- initialDelay
is the time to wait until the first execution happens
- delay
is the time to wait between 2 successive executions of the task
- unit
is the time unit used for the
initialDelay
and thedelay
parameters- r
is the callback to be executed
- returns
a cancelable that can be used to cancel the execution of this repeated task at any time.
- Definition Classes
- Base → Scheduler
- final def scheduleWithFixedDelay(initialDelay: FiniteDuration, delay: FiniteDuration)(action: => Unit): Cancelable
Schedules for execution a periodic task that is first executed after the given initial delay and subsequently with the given delay between the termination of one execution and the commencement of the next.
Schedules for execution a periodic task that is first executed after the given initial delay and subsequently with the given delay between the termination of one execution and the commencement of the next.
For example the following schedules a message to be printed to standard output every 10 seconds with an initial delay of 5 seconds:
val task = s.scheduleWithFixedDelay(5.seconds, 10.seconds) { print("Repeated message") } // later if you change your mind ... task.cancel()
- initialDelay
is the time to wait until the first execution happens
- delay
is the time to wait between 2 successive executions of the task
- action
is the callback to be executed
- returns
a cancelable that can be used to cancel the execution of this repeated task at any time.
- Definition Classes
- Scheduler
- def shutdown(): Unit
Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted.
Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted.
This method does not wait for previously submitted tasks to complete execution. Use awaitTermination(timeout:Long* to do that.
- Definition Classes
- TracingSchedulerService → SchedulerService
- final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def toString(): String
- Definition Classes
- AnyRef → Any
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- def withExecutionModel(em: ExecutionModel): TracingSchedulerService
Given a function that will receive the underlying ExecutionModel, returns a new Scheduler reference, based on the source, that exposes the transformed
ExecutionModel
when queried by means of the executionModel property.Given a function that will receive the underlying ExecutionModel, returns a new Scheduler reference, based on the source, that exposes the transformed
ExecutionModel
when queried by means of the executionModel property.This method enables reusing global scheduler references in a local scope, but with a slightly modified execution model to inject.
The contract of this method (things you can rely on):
- the source
Scheduler
must not be modified in any way - the implementation should wrap the source efficiently, such that the
result mirrors the source
Scheduler
in every way except for the execution model
Sample:
import monix.execution.Scheduler.global implicit val scheduler = { val em = global.executionModel global.withExecutionModel(em.withAutoCancelableLoops(true)) }
- Definition Classes
- TracingSchedulerService → SchedulerService → Scheduler
- the source
- def withUncaughtExceptionReporter(r: UncaughtExceptionReporter): TracingSchedulerService
Returns a new
Scheduler
that piggybacks on this, but uses the given exception reporter for reporting uncaught errors.Returns a new
Scheduler
that piggybacks on this, but uses the given exception reporter for reporting uncaught errors.Sample:
import monix.execution.Scheduler.global import monix.execution.UncaughtExceptionReporter implicit val scheduler = { val r = UncaughtExceptionReporter { e => e.printStackTrace() } global.withUncaughtExceptionReporter(r) }
- Definition Classes
- TracingSchedulerService → SchedulerService → Scheduler
Deprecated Value Members
- def finalize(): Unit
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.Throwable]) @Deprecated
- Deprecated
- def prepare(): ExecutionContext
- Definition Classes
- ExecutionContext
- Annotations
- @deprecated
- Deprecated
(Since version 2.12.0) preparation of ExecutionContexts will be removed
This is the API documentation for the Monix library.
Package Overview
monix.execution exposes lower level primitives for dealing with asynchronous execution:
Atomic
types, as alternative tojava.util.concurrent.atomic
monix.catnap exposes pure abstractions built on top of the Cats-Effect type classes:
monix.eval is for dealing with evaluation of results, thus exposing Task and Coeval.
monix.reactive exposes the
Observable
pattern:Observable
implementationsmonix.tail exposes Iterant for purely functional pull based streaming:
Batch
andBatchCursor
, the alternatives to Scala'sIterable
andIterator
respectively that we are using within Iterant's encodingYou can control evaluation with type you choose - be it Task, Coeval, cats.effect.IO or your own as long as you provide correct cats-effect or cats typeclass instance.