Returns the number of connected subscribers.

Note this might be an expensive operation.

Should be used for debugging purposes or for collecting metrics, but don't overuse because the accessed state is a volatile read, and counting subscribers might have linear complexity, depending on the underlying data-structure.

Characteristic function for an Observable instance, that creates the subscription and that eventually starts the streaming of events to the given Observer, being meant to be provided.

This function is "unsafe" to call because it does not protect the calls to the given Observer implementation in regards to unexpected exceptions that violate the contract, therefore the given instance must respect its contract and not throw any exceptions when the observable calls onNext, onComplete and onError. If it does, then the behavior is undefined.

Concatenates the source with another observable.

Ordering of subscription is preserved, so the second observable starts only after the source observable is completed successfully with an onComplete. On the other hand, the second observable is never subscribed if the source completes with an error.

Creates a new Observable that emits the given element and then it also emits the events of the source (prepend operation).

Creates a new Observable that emits the events of the source and then it also emits the given element (appended to the stream).

Given the source observable and another Observable, emits all of the items from the first of these Observables to emit an item and cancel the other.

Forces a buffered asynchronous boundary.

Internally it wraps the observer implementation given to onSubscribe into a BufferedSubscriber.

Normally Monix's implementation guarantees that events are not emitted concurrently, and that the publisher MUST NOT emit the next event without acknowledgement from the consumer that it may proceed, however for badly behaved publishers, this wrapper provides the guarantee that the downstream Observer given in subscribe will not receive concurrent events.

WARNING: if the buffer created by this operator is unbounded, it can blow up the process if the data source is pushing events faster than what the observer can consume, as it introduces an asynchronous boundary that eliminates the back-pressure requirements of the data source. Unbounded is the default overflowStrategy, see OverflowStrategy for options.

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers). The underlying subject used is a BehaviorSubject.

Buffers signals while busy, after which it emits the buffered events as a single bundle.

This operator starts applying back-pressure when the underlying buffer's size is exceeded.

Returns an observable that emits buffers of items it collects from the source observable. The resulting observable emits buffers every skip items, each containing count items.

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

For count and skip there are 3 possibilities:

1. in case skip == count, then there are no items dropped and no overlap, the call being equivalent to buffer(count)
2. in case skip < count, then overlap between buffers happens, with the number of elements being repeated being count - skip
3. in case skip > count, then skip - count elements start getting dropped between windows

Periodically gather items emitted by an observable into bundles and emit these bundles rather than emitting the items one at a time.

This version of buffer emits a new bundle of items periodically, every timespan amount of time, containing all items emitted by the source Observable since the previous bundle emission.

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

Periodically gather items emitted by an observable into bundles and emit these bundles rather than emitting the items one at a time.

The resulting observable emits connected, non-overlapping buffers, each of a fixed duration specified by the timespan argument or a maximum size specified by the maxCount argument (whichever is reached first).

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

Periodically gather items emitted by an observable into bundles and emit these bundles rather than emitting the items one at a time. Back-pressure the source when the buffer is full.

The resulting observable emits connected, non-overlapping buffers, each of a fixed duration specified by the period argument.

The bundles are emitted at a fixed rate. If the source is silent, then the resulting observable will start emitting empty sequences.

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

A maxSize argument is specified as the capacity of the bundle. In case the source is too fast and maxSize is reached, then the source will be back-pressured.

The difference with bufferTimedAndCounted is that bufferTimedWithPressure applies back-pressure from the time when the buffer is full until the buffer is emitted, whereas bufferTimedAndCounted will forcefully emit the buffer when it's full.

Periodically gather items emitted by an observable into bundles and emit these bundles rather than emitting the items one at a time. This version of buffer is emitting items once the internal buffer has reached the given count.

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

Periodically gather items emitted by an observable into bundles and emit these bundles rather than emitting the items one at a time, whenever the selector observable signals an event.

The resulting observable collects the elements of the source in a buffer and emits that buffer whenever the given selector observable emits an onNext event, when the buffer is emitted as a sequence downstream and then reset. Thus the resulting observable emits connected, non-overlapping bundles triggered by the given selector.

If selector terminates with an onComplete, then the resulting observable also terminates normally. If selector terminates with an onError, then the resulting observable also terminates with an error.

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

A maxSize argument is specified as the capacity of the bundle. In case the source is too fast and maxSize is reached, then the source will be back-pressured.

Periodically gather items emitted by an observable into bundles and emit these bundles rather than emitting the items one at a time, whenever the selector observable signals an event.

The resulting observable collects the elements of the source in a buffer and emits that buffer whenever the given selector observable emits an onNext event, when the buffer is emitted as a sequence downstream and then reset. Thus the resulting observable emits connected, non-overlapping bundles triggered by the given selector.

If selector terminates with an onComplete, then the resulting observable also terminates normally. If selector terminates with an onError, then the resulting observable also terminates with an error.

If the source observable completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

Caches the emissions from the source Observable and replays them in order to any subsequent Subscribers. This operator has similar behavior to replay except that this auto-subscribes to the source Observable rather than returning a ConnectableObservable for which you must call connect to activate the subscription.

When you call cache, it does not yet subscribe to the source Observable and so does not yet begin caching items. This only happens when the first Subscriber calls the resulting Observable's subscribe method.

Caches the emissions from the source Observable and replays them in order to any subsequent Subscribers. This operator has similar behavior to replay except that this auto-subscribes to the source Observable rather than returning a ConnectableObservable for which you must call connect to activate the subscription.

When you call cache, it does not yet subscribe to the source Observable and so does not yet begin caching items. This only happens when the first Subscriber calls the resulting Observable's subscribe method.

Note: You sacrifice the ability to cancel the origin when you use the cache operator so be careful not to use this on Observables that emit an infinite or very large number of items that will use up memory.

Applies the given partial function to the source for each element for which the given partial function is defined.

Creates a new observable from the source and another given observable, by emitting elements combined in pairs. If one of the observables emits fewer events than the other, then the rest of the unpaired events are ignored.

See zip for an alternative that pairs the items in strict sequence.

Creates a new observable from the source and another given observable, by emitting elements combined in pairs. If one of the observables emits fewer events than the other, then the rest of the unpaired events are ignored.

See zipMap for an alternative that pairs the items in strict sequence.

Ignores all items emitted by the source Observable and only calls onCompleted or onError.

Creates a new Task that will consume the source observable and upon completion of the source it will complete with Unit.

Concatenates the sequence of observables emitted by the source into one observable, without any transformation.

You can combine the items emitted by multiple observables so that they act like a single sequence by using this operator.

The difference between the concat operation and mergeis that concat cares about the ordering of sequences (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, concat is safe to use in all contexts, whereas merge requires buffering.

Concatenates the sequence of observables emitted by the source into one observable, without any transformation.

You can combine the items emitted by multiple observables so that they act like a single sequence by using this operator.

The difference between the concat operation and mergeis that concat cares about the ordering of sequences (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, concat is safe to use in all contexts, whereas merge requires buffering.

This version is reserving onError notifications until all of the observables complete and only then passing the issued errors(s) downstream. Note that the streamed error is a CompositeException, since multiple errors from multiple streams can happen.

Applies a function that you supply to each item emitted by the source observable, where that function returns observables, and then concatenating those resulting sequences and emitting the results of this concatenation.

The difference between the concat operation and mergeis that concat cares about the ordering of sequences (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, concat is safe to use in all contexts, whereas merge requires buffering.

Applies a function that you supply to each item emitted by the source observable, where that function returns sequences and then concatenating those resulting sequences and emitting the results of this concatenation.

This version is reserving onError notifications until all of the observables complete and only then passing the issued errors(s) downstream. Note that the streamed error is a CompositeException, since multiple errors from multiple streams can happen.

On execution, consumes the source observable with the given Consumer, effectively transforming the source observable into a Task.

Creates a new Observable that emits the total number of onNext events that were emitted by the source.

Note that this Observable emits only one item after the source is complete. And in case the source emits an error, then only that error will be emitted.

Creates a task that emits the total number of onNext events that were emitted by the source.

Only emit an item from an observable if a particular timespan has passed without it emitting another item.

Note: If the source observable keeps emitting items more frequently than the length of the time window, then no items will be emitted by the resulting observable.

Emits the last item from the source Observable if a particular timespan has passed without it emitting another item, and keeps emitting that item at regular intervals until the source breaks the silence.

So compared to regular debounceTo this version keeps emitting the last item of the source.

Note: If the source Observable keeps emitting items more frequently than the length of the time window then no items will be emitted by the resulting Observable.

Doesn't emit anything until a timeout period passes without the source emitting anything. When that timeout happens, we subscribe to the observable generated by the given function, an observable that will keep emitting until the source will break the silence by emitting another event.

Note: If the source observable keeps emitting items more frequently than the length of the time window, then no items will be emitted by the resulting Observable.

Emit items from the source, or emit a default item if the source completes after emitting no items.

Delays emitting the final onComplete event by the specified amount.

Returns an Observable that emits the items emitted by the source Observable shifted forward in time by a specified delay.

Each time the source Observable emits an item, delay starts a timer, and when that timer reaches the given duration, the Observable returned from delay emits the same item.

NOTE: this delay refers strictly to the time between the onNext event coming from our source and the time it takes the downstream observer to get this event. On the other hand the operator is also applying back-pressure, so on slow observers the actual time passing between two successive events may be higher than the specified duration.

Returns an Observable that emits the items emitted by the source Observable shifted forward in time.

This variant of delay sets its delay duration on a per-item basis by passing each item from the source Observable into a function that returns an Observable and then monitoring those Observables. When any such Observable emits an item or completes, the Observable returned by delay emits the associated item.

Hold an Observer's subscription request for a specified amount of time before passing it on to the source Observable.

Hold an Observer's subscription request until the given trigger observable either emits an item or completes, before passing it on to the source Observable.

If the given trigger completes in error, then the subscription is terminated with onError.

Converts the source Observable that emits Notification[A] (the result of materialize) back to an Observable that emits A.

Suppress the duplicate elements emitted by the source Observable.

WARNING: this requires unbounded buffering.

Given a function that returns a key for each element emitted by the source Observable, suppress duplicates items.

WARNING: this requires unbounded buffering.

Suppress duplicate consecutive items emitted by the source Observable

Suppress duplicate consecutive items emitted by the source Observable

Executes the given callback just after the subscription happens.

Executes the given callback after the stream has ended either with an onComplete or onError event, or when the streaming stops by a downstream Stop being signaled.

This differs from doOnTerminate in that this happens *after* the onComplete or onError notification.

Evaluates the task generated by the given callback after the stream has ended either with an onComplete or onError event, or when the streaming stops by a downstream Stop being signaled.

This operation subsumes doOnEarlyStopEval and the callback-generated Task will back-pressure the source when applied for Stop events returned by onNext and thus the upstream source will receive the Stop result only after the task has finished executing.

This differs from doOnTerminateEval in that this happens *after* the onComplete or onError notification.

Executes the given callback when the stream has ended with an onComplete event, but before the complete event is emitted.

Unless you know what you're doing, you probably want to use doOnTerminate and doOnSubscriptionCancel for proper disposal of resources on completion.

Evaluates the given task when the stream has ended with an onComplete event, but before the complete event is emitted.

The task gets evaluated and is finished *before* the onComplete signal gets sent downstream.

Unless you know what you're doing, you probably want to use doOnTerminateEval and doOnSubscriptionCancel for proper disposal of resources on completion.

Executes the given callback when the streaming is stopped due to a downstream Stop signal returned by onNext.

Executes the given task when the streaming is stopped due to a downstream Stop signal returned by onNext.

The given task gets evaluated *before* the upstream receives the Stop event (is back-pressured).

Executes the given callback when the stream is interrupted with an error, before the onError event is emitted downstream.

NOTE: should protect the code in this callback, because if it throws an exception the onError event will prefer signaling the original exception and otherwise the behavior is undefined.

Executes the given task when the stream is interrupted with an error, before the onError event is emitted downstream.

NOTE: should protect the code in this callback, because if it throws an exception the onError event will prefer signaling the original exception and otherwise the behavior is undefined.

Executes the given callback for each element generated by the source Observable, useful for doing side-effects.

Executes the given callback on each acknowledgement received from the downstream subscriber.

This method helps in executing logic after messages get processed, for example when messages are polled from some distributed message queue and an acknowledgement needs to be sent after each message in order to mark it as processed.

Executes the given callback on each acknowledgement received from the downstream subscriber, executing a generated Task and back-pressuring until the task is done.

This method helps in executing logic after messages get processed, for example when messages are polled from some distributed message queue and an acknowledgement needs to be sent after each message in order to mark it as processed.

Evaluates the given callback for each element generated by the source Observable, useful for triggering async side-effects.

Executes the given callback only for the first element generated by the source Observable, useful for doing a piece of computation only when the stream starts.

Executes the given callback just before the subscription happens.

Executes the given callback when the connection is being cancelled.

Executes the given callback right before the streaming is ended either with an onComplete or onError event, or when the streaming stops by a downstream Stop being signaled.

It is the equivalent of calling:

• doOnComplete
• doOnError
• doOnEarlyStop

This differs from doAfterTerminate in that this happens *before* the onComplete or onError notification.

Evaluates the task generated by the given callback right before the streaming is ended either with an onComplete or onError event, or when the streaming stops by a downstream Stop being signaled.

The callback-generated Task will back-pressure the source when applied for Stop events returned by onNext and thus the upstream source will receive the Stop result only after the task has finished executing.

It is the equivalent of calling:

• doOnCompleteEval
• doOnErrorEval
• doOnEarlyStopEval

This differs from doAfterTerminateEval in that this happens *before* the onComplete or onError notification.

Drops the first n elements (from the start).

Creates a new observable that drops the events of the source, only for the specified timestamp window.

Drops the last n elements (from the end).

Discard items emitted by the source until a second observable emits an item or completes.

If the trigger observable completes in error, then the resulting observable will also end in error when it notices it (next time an element is emitted by the source).

Drops the longest prefix of elements that satisfy the given predicate and returns a new observable that emits the rest.

Drops the longest prefix of elements that satisfy the given function and returns a new observable that emits the rest. In comparison with dropWhile, this version accepts a function that takes an additional parameter: the zero-based index of the element.

Utility that can be used for debugging purposes.

Mirror the source observable as long as the source keeps emitting items, otherwise if timeout passes without the source emitting anything new then the observable will emit the last item.

This is the rough equivalent of:

Observable.merge(source, source.debounce(period))

Note: If the source Observable keeps emitting items more frequently than the length of the time window then the resulting observable will mirror the source exactly.

Mirror the source observable as long as the source keeps emitting items, otherwise if timeout passes without the source emitting anything new then the observable will start emitting the last item repeatedly.

Note: If the source Observable keeps emitting items more frequently than the length of the time window then the resulting observable will mirror the source exactly.

Creates a new Observable that emits the events of the source and then it also emits the given elements (appended to the stream).

Emits the given exception instead of onComplete.

Specify an override for the Scheduler that will be used for subscribing and for observing the source.

Normally the Scheduler gets injected implicitly when doing subscribe, but this operator overrides the injected subscriber for the given source. And if the source is normally using that injected scheduler (given by subscribe), then the effect will be that all processing will now happen on the override.

To put it in other words, in Monix it's usually the consumer and not the producer that specifies the scheduler and this operator allows for a different behavior.

This operator also includes the effects of subscribeOn, meaning that the subscription logic itself will start on the provided scheduler.

IMPORTANT: This operator is a replacement for the observeOn operator from ReactiveX, but does not work in the same way. The observeOn operator forces the signaling to happen on a given Scheduler, but executeOn is more relaxed, usage is not forced, the source just gets injected with a different scheduler and it's up to the source to actually use it. This also means the effects are more far reaching, because the whole chain until the call of this operator is affected.

Alias for Observable.fork(fa, scheduler).

Mirrors the source observable, but upon subscription ensure that the evaluation forks into a separate (logical) thread.

The execution is managed by the injected scheduler in subscribe().

Alias for Observable.fork(fa).

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

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

observable.executeWithModel(AlwaysAsyncExecution)

Returns an Observable which emits a single value, either true, in case the given predicate holds for at least one item, or false otherwise.

Returns a Task which emits either true, in case the given predicate holds for at least one item, or false otherwise.

Returns an observable that emits a single Throwable, in case an error was thrown by the source, otherwise it isn't going to emit anything.

Only emits those items for which the given predicate holds.

Returns an Observable which only emits the first item for which the predicate holds.

Returns a task which emits the first item for which the predicate holds.

Creates a new Task that upon execution will signal the first generated element of the source observable.

In case the stream was empty, then the Task gets completed in error with a NoSuchElementException.

Creates a new Task that upon execution will signal the first generated element of the source observable.

Returns an Option because the source can be empty.

Emits the first element emitted by the source, or otherwise if the source is completed without emitting anything, then the default is emitted.

Alias for headOrElse.

Creates a new Task that upon execution will signal the first generated element of the source observable.

In case the stream was empty, then the given default gets evaluated and emitted.

Applies a function that you supply to each item emitted by the source observable, where that function returns sequences that can be observed, and then concatenating those resulting sequences and emitting the results of this concatenation.

Alias for concatMap.

The difference between the concat operation and mergeis that concat cares about the ordering of sequences (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, concat is safe to use in all contexts, whereas merge requires buffering.

Applies a function that you supply to each item emitted by the source observable, where that function returns sequences and then concatenating those resulting sequences and emitting the results of this concatenation.

It's an alias for concatMapDelayErrors.

An alias of switchMap.

Returns a new observable that emits the items emitted by the observable most recently generated by the mapping function.

Applies a binary operator to a start value and to elements produced by the source observable, going from left to right, producing and concatenating observables along the way.

It's the combination between scan and flatMap.

Applies a binary operator to a start value and to elements produced by the source observable, going from left to right, producing and concatenating observables along the way.

This version of flatScan delays all errors until onComplete, when it will finally emit a CompositeException. It's the combination between scan and flatMapDelayErrors.

Concatenates the sequence of observables emitted by the source into one observable, without any transformation.

You can combine the items emitted by multiple observables so that they act like a single sequence by using this operator.

The difference between the concat operation and mergeis that concat cares about the ordering of sequences (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, concat is safe to use in all contexts, whereas merge requires buffering.

Alias for concat.

Alias for concatDelayErrors.

Concatenates the sequence of observables emitted by the source into one observable, without any transformation.

You can combine the items emitted by multiple observables so that they act like a single sequence by using this operator.

The difference between the concat operation and mergeis that concat cares about the ordering of sequences (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, concat is safe to use in all contexts, whereas merge requires buffering.

This version is reserving onError notifications until all of the observables complete and only then passing the issued errors(s) downstream. Note that the streamed error is a CompositeException, since multiple errors from multiple streams can happen.

Alias for switch

Convert an observable that emits observables into a single observable that emits the items emitted by the most-recently-emitted of those observables.

Applies a binary operator to a start value and all elements of this Observable, going left to right and returns a new Observable that emits only one item before onComplete.

Applies a binary operator to a start value and all elements of the source, going left to right and returns a new Task that upon evaluation will eventually emit the final result.

Folds the source observable, from start to finish, until the source completes, or until the operator short-circuits the process by returning false.

Note that a call to foldLeftF is equivalent to this function being called with an operator always returning true as the first member of its result.

Folds the source observable, from start to finish, until the source completes, or until the operator short-circuits the process by returning false.

Note that a call to foldLeftL is equivalent to this function being called with an operator always returning true as the first member of its result.

Returns an Observable that emits a single boolean, either true, in case the given predicate holds for all the items emitted by the source, or false in case at least one item is not verifying the given predicate.

Returns a Task that emits a single boolean, either true, in case the given predicate holds for all the items emitted by the source, or false in case at least one item is not verifying the given predicate.

Subscribes to the source Observable and foreach element emitted by the source it executes the given callback.

Creates a new Task that will consume the source observable, executing the given callback for each element.

Groups the items emitted by an Observable according to a specified criterion, and emits these grouped items as GroupedObservables, one GroupedObservable per group.

Note: A GroupedObservable will cache the items it is to emit until such time as it is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those GroupedObservables that do not concern you. Instead, you can signal to them that they may discard their buffers by doing something like source.take(0).

Only emits the first element emitted by the source observable, after which it's completed immediately.

Alias for firstL.

Alias for firstOptionL.

Emits the first element emitted by the source, or otherwise if the source is completed without emitting anything, then the default is emitted.

Alias for firstOrElseL.

Alias for completed. Ignores all items emitted by the source and only calls onCompleted or onError.

Creates a new observable from this observable and another given observable by interleaving their items into a strictly alternating sequence.

So the first item emitted by the new observable will be the item emitted by self, the second item will be emitted by the other observable, and so forth; when either self or other calls onCompletes, the items will then be directly coming from the observable that has not completed; when onError is called by either self or other, the new observable will call onError and halt.

See merge for a more relaxed alternative that doesn't emit items in strict alternating sequence.

Returns an Observable that emits true if the source Observable is empty, otherwise false.

Returns a task that emits true if the source observable is empty, otherwise false.

Only emits the last element emitted by the source observable, after which it's completed immediately.

Returns a Task that upon execution will signal the last generated element of the source observable.

In case the stream was empty, then the Task gets completed in error with a NoSuchElementException.

Returns a Task that upon execution will signal the last generated element of the source observable.

Returns an Option because the source can be empty.

Creates a new Task that upon execution will signal the last generated element of the source observable.

In case the stream was empty, then the given default gets evaluated and emitted.

Transforms the source using the given operator.

Returns a new observable that applies the given function to each item emitted by the source and emits the result.

Given a mapping function that maps events to tasks, applies it in parallel on the source, but with a specified parallelism, which indicates the maximum number of tasks that can be executed in parallel.

Similar in spirit with Consumer.loadBalance, but expressed as an operator that executes Task instances in parallel.

Note that when the specified parallelism is 1, it has the same behavior as mapTask.

Alias for mapTask.

Maps elements from the source using a function that can do asynchronous processing by means of scala.concurrent.Future.

Given a source observable, this function is basically the equivalent of doing:

observable.concatMap(a => Observable.fromFuture(f(a)))

However prefer this operator to concatMap because it is more clear and has better performance.

Maps elements from the source using a function that can do asynchronous processing by means of Task.

Given a source observable, this function is basically the equivalent of doing:

observable.concatMap(a => Observable.fromTask(f(a)))

However prefer this operator to concatMap because it is more clear and has better performance.

Converts the source Observable that emits A into an Observable that emits Notification[A].

Takes the elements of the source Observable and emits the element that has the maximum key value, where the key is generated by the given function f.

Takes the elements of the source and emits the element that has the maximum key value, where the key is generated by the given function f.

Takes the elements of the source Observable and emits the maximum value, after the source has completed.

Takes the elements of the source and emits the maximum value, after the source has completed.

Merges the sequence of Observables emitted by the source into one Observable, without any transformation.

You can combine the items emitted by multiple Observables so that they act like a single Observable by using this operator.

Merges the sequence of Observables emitted by the source into one Observable, without any transformation.

You can combine the items emitted by multiple Observables so that they act like a single Observable by using this operator.

This version is reserving onError notifications until all of the observables complete and only then passing the issued errors(s) downstream. Note that the streamed error is a CompositeException, since multiple errors from multiple streams can happen.

Creates a new observable by applying a function that you supply to each item emitted by the source observable, where that function returns an observable, and then merging those resulting observable and emitting the results of this merger.

The difference between this and concatMap is that concatMap cares about ordering of emitted items (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, the concat operation is safe to use in all contexts, whereas merge requires buffering.

Creates a new observable by applying a function that you supply to each item emitted by the source observable, where that function returns an observable, and then merging those resulting observable and emitting the results of this merger.

The difference between this and concatMap is that concatMap cares about ordering of emitted items (e.g. all items emitted by the first observable in the sequence will come before the elements emitted by the second observable), whereas merge doesn't care about that (elements get emitted as they come). Because of back-pressure applied to observables, the concat operation is safe to use in all contexts, whereas merge requires buffering.

This version is reserving onError notifications until all of the observables complete and only then passing the issued errors(s) downstream. Note that the streamed error is a CompositeException, since multiple errors from multiple streams can happen.

Takes the elements of the source Observable and emits the element that has the minimum key value, where the key is generated by the given function f.

Takes the elements of the source and emits the element that has the minimum key value, where the key is generated by the given function f.

Takes the elements of the source Observable and emits the minimum value, after the source has completed.

Takes the elements of the source and emits the minimum value, after the source has completed.

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers).

Returns an Observable that emits false if the source Observable is empty, otherwise true.

Returns a task that emits false if the source observable is empty, otherwise true.

Operator that specifies a different Scheduler, on which subscribers will observe events, instead of the default one.

This overloaded version of observeOn takes an extra OverflowStrategy parameter specifying the behavior of the underlying buffer.

Operator that specifies a different Scheduler, on which subscribers will observe events, instead of the default one.

An Observable with an applied observeOn call will forward events into a buffer that uses the specified Scheduler reference to cycle through events and to make onNext calls to downstream listeners.

Example:

import monix.execution.Scheduler
val io = Scheduler.io("my-io")

source.map(_ + 1)
  .observeOn(io)
  .foreach(x => println(x))

In the above example the first map (whatever comes before the observeOn call) gets executed using the default Scheduler (might execute on the current thread even), however the foreach that's specified after observeOn will get executed on the indicated Scheduler.

NOTE: this operator does not guarantee that downstream listeners will actually use the specified Scheduler to process events, because this depends on the rest of the pipeline. E.g. this will not work OK:

source.observeOn(io).asyncBoundary(Unbounded)

This sample might not do what a user of observeOn would want. Indeed the implementation will use the provided io reference for calling onNext / onComplete / onError events, however because of the following asynchronous boundary created the actual listeners will probably end up being execute on a different Scheduler.

The underlying implementation uses a buffer to forward events. The OverflowStrategy being applied is the default one.

If the connection is cancelled then trigger a CancellationException.

A connection can be cancelled with the help of the Cancelable returned on subscribe.

Because the cancellation is effectively concurrent with the signals the Observer receives and because we need to uphold the contract, this operator will effectively synchronize access to onNext, onComplete and onError. It will also watch out for asynchronous Stop events.

In other words, this operator does heavy synchronization, can prove to be inefficient and you should avoid using it because the signaled error can interfere with functionality from other operators that use cancellation internally and cancellation in general is a side-effecting operation that should be avoided, unless it's necessary.

Returns an Observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case the streaming of events continues with the specified backup sequence.

The created Observable mirrors the behavior of the source in case the source does not end with an error.

NOTE that compared with onErrorResumeNext from Rx.NET, the streaming is not resumed in case the source is terminated normally with an onComplete.

Returns an observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case the streaming of events fallbacks to an observable emitting a single element generated by the backup function.

See onErrorRecover for the version that takes a partial function as a parameter.

Returns an Observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case the streaming of events continues with the specified backup sequence generated by the given function.

See onErrorRecoverWith for the version that takes a partial function as a parameter.

Returns an observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case the streaming of events fallbacks to an observable emitting a single element generated by the backup function.

The created Observable mirrors the behavior of the source in case the source does not end with an error or if the thrown Throwable is not matched.

See onErrorHandle for the version that takes a total function as a parameter.

Returns an Observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case the streaming of events continues with the specified backup sequence generated by the given function.

The created Observable mirrors the behavior of the source in case the source does not end with an error or if the thrown Throwable is not matched.

See onErrorHandleWith for the version that takes a total function as a parameter.

Returns an Observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case it tries subscribing to the source again in the hope that it will complete without an error.

The number of retries is limited by the specified maxRetries parameter, so for an Observable that always ends in error the total number of subscriptions that will eventually happen is maxRetries + 1.

Returns an Observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case it tries subscribing to the source again in the hope that it will complete without an error.

The given predicate establishes if the subscription should be retried or not.

Returns an Observable that mirrors the behavior of the source, unless the source is terminated with an onError, in which case it tries subscribing to the source again in the hope that it will complete without an error.

NOTE: The number of retries is unlimited, so something like Observable.error(new RuntimeException).onErrorRestartUnlimited will loop forever.

Given a Pipe, transform the source observable with it.

Returns an observable that emits the results of invoking a specified selector on items emitted by a ConnectableObservable, which shares a single subscription to the underlying sequence.

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers). The underlying subject used is a PublishSubject.

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers). The underlying subject used is a AsyncSubject.

Returns an observable that emits the results of invoking a specified selector on items emitted by a ConnectableObservable, which shares a single subscription to the underlying sequence.

Applies a binary operator to a start value and all elements of this Observable, going left to right and returns a new Observable that emits only one item before onComplete.

Repeats the items emitted by the source continuously. It caches the generated items until onComplete and repeats them forever.

It terminates either on error or if the source is empty.

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers). The underlying subject used is a ReplaySubject.

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers). The underlying subject used is a ReplaySubject.

Keeps restarting / resubscribing the source until the predicate returns true for the the first emitted element, after which it starts mirroring the source.

Creates a new CancelableFuture that upon execution will signal the last generated element of the source observable. Returns an Option because the source can be empty.

Creates a new CancelableFuture that upon execution will signal the last generated element of the source observable. Returns an Option because the source can be empty.

Emit the most recent items emitted by the source within periodic time intervals.

Use the sample operator to periodically look at an observable to see what item it has most recently emitted since the previous sampling. Note that if the source observable has emitted no items since the last time it was sampled, the observable that results from the sample operator will emit no item for that sampling period.

Returns an observable that, when the specified sampler emits an item or completes, emits the most recently emitted item (if any) emitted by the source since the previous emission from the sampler.

Use the sampleBy operator to periodically look at an observable to see what item it has most recently emitted since the previous sampling. Note that if the source observable has emitted no items since the last time it was sampled, the observable that results from the sampleBy operator will emit no item.

Emit the most recent items emitted by an observable within periodic time intervals. If no new value has been emitted since the last time it was sampled, it signals the last emitted value anyway.

Returns an observable that, when the specified sampler observable emits an item or completes, emits the most recently emitted item (if any) emitted by the source Observable since the previous emission from the sampler observable. If no new value has been emitted since the last time it was sampled, it signals the last emitted value anyway.

Applies a binary operator to a start value and all elements of this Observable, going left to right and returns a new Observable that emits on each step the result of the applied function.

Similar to foldLeftF, but emits the state on each step. Useful for modeling finite state machines.

Returns a new Observable that multi-casts (shares) the original Observable.

Creates a new Observable that emits the given elements and then it also emits the events of the source (prepend operation).

Subscribes to the stream.

Subscribes to the stream.

Subscribes to the stream.

Subscribes to the stream.

Subscribes to the stream.

Subscribes to the stream.

Returns a new Observable that uses the specified Scheduler for initiating the subscription.

Given a source that emits numeric values, the sum operator sums up all values and at onComplete it emits the total.

Given a source that emits numeric values, the sum operator sums up all values and returns the result.

Convert an observable that emits observables into a single observable that emits the items emitted by the most-recently-emitted of those observables.

In case the source is empty, switch to the given backup.

Returns a new observable that emits the items emitted by the observable most recently generated by the mapping function.

Drops the first element of the source observable, emitting the rest.

Selects the first n elements (from the start).

Creates a new Observable that emits the events of the source, only for the specified timestamp, after which it completes.

Creates a new Observable that emits every n-th event from the source, dropping intermediary events.

Creates a new observable that only emits the last n elements emitted by the source.

In case the source triggers an error, then the underlying buffer gets dropped and the error gets emitted immediately.

Creates a new observable that mirrors the source until the given trigger emits either an element or onComplete, after which it is completed.

The resulting observable is completed as soon as trigger emits either an onNext or onComplete. If trigger emits an onError, then the resulting observable is also completed with error.

Takes longest prefix of elements that satisfy the given predicate and returns a new Observable that emits those elements.

Takes longest prefix of elements that satisfy the given predicate and returns a new Observable that emits those elements.

Returns an Observable that emits only the first item emitted by the source Observable during sequential time windows of a specified duration.

This differs from Observable!.throttleLast in that this only tracks passage of time whereas throttleLast ticks at scheduled intervals.

Emit the most recent items emitted by the source within periodic time intervals.

Alias for sample.

Only emit an item from an observable if a particular timespan has passed without it emitting another item.

Note: If the source observable keeps emitting items more frequently than the length of the time window, then no items will be emitted by the resulting observable.

Alias for debounce.

Returns an observable that mirrors the source but that will trigger a DownstreamTimeoutException in case the downstream subscriber takes more than the given timespan to process an onNext message.

Note that this ignores the time it takes for the upstream to send onNext messages. For detecting slow producers see timeoutOnSlowUpstream.

Returns an observable that mirrors the source but applies a timeout for each emitted item by the upstream. If the next item isn't emitted within the specified timeout duration starting from its predecessor, the resulting Observable terminates and notifies observers of a TimeoutException.

Note that this ignores the time it takes to process onNext. If dealing with a slow consumer, see timeoutOnSlowDownstream.

Returns an observable that mirrors the source but applies a timeout for each emitted item by the upstream. If the next item isn't emitted within the specified timeout duration starting from its predecessor, the source is terminated and the downstream gets subscribed to the given backup.

Note that this ignores the time it takes to process onNext. If dealing with a slow consumer, see timeoutOnSlowDownstream.

Returns a Task that upon evaluation will collect all items from the source in a Scala List and return this list instead.

WARNING: for infinite streams the process will eventually blow up with an out of memory error.

Wraps this Observable into a org.reactivestreams.Publisher. See the Reactive Streams protocol that Monix implements.

Transforms the source using the given transformer function.

Makes the source Observable uninterruptible such that a cancel signal has no effect.

val cancelable = Observable
  .eval(println("Hello!"))
  .delayExecution(10.seconds)
  .subscribe()

// No longer works
cancelable.cancel()

// After 10 seconds
//=> Hello!

Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e. whose source is shared by all observers).

This operator is unsafe because Subject objects are stateful and have to obey the Observer contract, meaning that they shouldn't be subscribed multiple times, so they are error prone. Only use if you know what you're doing, otherwise prefer the safe multicast operator.

While the destination observer is busy, buffers events, applying the given overflowStrategy.

While the destination observer is busy, drop the incoming events.

While the destination observer is busy, drop the incoming events. When the downstream recovers, we can signal a special event meant to inform the downstream observer how many events where dropped.

Combines the elements emitted by the source with the latest element emitted by another observable.

Similar with combineLatest, but only emits items when the single source emits an item (not when any of the Observables that are passed to the operator do, as combineLatest does).

Combines the elements emitted by the source with the latest elements emitted by two observables.

Similar with combineLatest, but only emits items when the single source emits an item (not when any of the Observables that are passed to the operator do, as combineLatest does).

Combines the elements emitted by the source with the latest elements emitted by three observables.

Similar with combineLatest, but only emits items when the single source emits an item (not when any of the Observables that are passed to the operator do, as combineLatest does).

Combines the elements emitted by the source with the latest elements emitted by four observables.

Similar with combineLatest, but only emits items when the single source emits an item (not when any of the Observables that are passed to the operator do, as combineLatest does).