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

UNSAFE PROTOCOL: This function is "unsafe" to call because it does not protect the calls to the given Observer implementation and thus knowledge of the protocol is needed.

Prefer normal subscribe when consuming a stream, these unsafe subscription methods being useful when building operators and for testing purposes.

Normal subscribe protects users in these ways:

• it does a best effort attempt to catch and report exceptions that violate the protocol
• the final onComplete or onError message is guaranteed to be signaled after the completion of the acknowledgement received from the last onNext; the internal protocol doesn't require back-pressuring of this last message for performance reasons

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

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.

Visual Example

streamA: a1 -- -- a2 -- -- a3 -- a4 -- --
streamB: b1 -- -- b2 -- b3 -- -- -- -- b4

result: a1, a2, a3, a4, b1, b2, b3, b4

Forces a buffered asynchronous boundary. Asynchronous boundary refers to an independent processing of an upstream and a downstream - producer does not have to wait for consumer to acknowledge a new event.

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.

Usage:

import monix.eval.Task
import scala.concurrent.duration._

Observable("A", "B", "C", "D")
  .mapEval(i => Task { println(s"1: Processing $$i"); i ++ i })
  .asyncBoundary(OverflowStrategy.Unbounded)
  .mapEval(i => Task { println(s"2: Processing $$i") }.delayExecution(100.millis))

// Without asyncBoundary it would process A, AA, B, BB, ...
// 1: Processing A
// 1: Processing B
// 1: Processing C
// 1: Processing D
// 2: Processing AA
// 2: Processing BB
// 2: Processing CC
// 2: Processing DD

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.

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

Implementation of bracket from cats.effect.Bracket.

See documentation.

Implementation of bracketCase from cats.effect.Bracket.

See documentation.

Version of bracketCase that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So in release you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Version of bracket that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So in release you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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.

Usage:

import monix.eval.Task
import scala.concurrent.duration._

Observable.range(1, 6)
  .doOnNext(l => Task(println(s"Started $$l")))
  .bufferIntrospective(maxSize = 2)
  .doOnNext(l => Task(println(s"Emitted batch $$l")))
  .mapEval(l => Task(println(s"Processed batch $$l")).delayExecution(500.millis))

// Started 1
// Emitted batch List(1)
// Started 2
// Started 3
// Processed batch List(1)
// Emitted batch List(2, 3)
// Started 4
// Started 5
// Processed batch List(2, 3)
// Emitted batch List(4, 5)
// Processed batch List(4, 5)

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 bufferTumbling(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

Usage:

// Emits [2, 3], [5, 6]
Observable.range(2, 7)
  .bufferSliding(count = 2, skip = 3)

// Emits [2, 3, 4], [4, 5, 6]
Observable.range(2, 7)
  .bufferSliding(count = 3, skip = 2)

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.

A sizeOf argument is specified as the weight each element represents in the bundle. Defaults to count each element as weighting 1.

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.

Usage:

// Emits [2, 3], [4, 5], [6]
Observable.range(2, 7)
  .bufferTumbling(count = 2)

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.

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

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.

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

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

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

Creates a new observable from the source and another given observable, by emitting elements combined in pairs.

It emits an item whenever any of the source Observables emits an item (so long as each of the source Observables has emitted at least one item).

Visual Example

stream1: 1 - - 2 - - 3 - 4 - -
stream2: 1 - - 2 - 3 - - - - 4

result: (1, 1), (2, 2), (2, 3), (3, 3), (4, 3), (4, 4)

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.

It emits an item whenever any of the source Observables emits an item (so long as each of the source Observables has emitted at least one item).

Visual Example

stream1: 1 - - 2 - - 3 - 4 - -
stream2: 1 - - 2 - 3 - - - - 4

result: (1, 1), (2, 2), (2, 3), (3, 3), (4, 3), (4, 4)

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.

Polymorphic version of completedL that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLift conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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.

This operation is the "monadic bind", implementing the flatMap operation of cats.Monad.

Concat vs Merge

The difference between the concat operation and merge is 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. Or in other words concat has deterministic, lawful behavior (being the "monadic bind"), whereas merge has non-deterministic behavior.

Equivalence with concatMap

The concat operation is basically concatMap with the identity function, as you can count on this equivalence:

stream.concat <-> stream.concatMap(x => x)

Visual Example

streamA: a1 -- -- a2 -- -- a3 -- a4 -- --
streamB: b1 -- -- b2 -- b3 -- -- -- -- b4

result: a1, a2, a3, a4, b1, b2, b3, b4

Version of concat that delays errors emitted by child observables until the stream completes.

Delaying Errors

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.

Example

val dummy1 = new RuntimeException("dummy1")
val dummy2 = new RuntimeException("dummy2")

val stream = Observable(
  Observable(1).endWithError(dummy1),
  Observable.raiseError(dummy2),
  Observable(2, 3)
)

val concatenated =
  stream.concatDelayErrors

The resulting stream in this example emits 1, 2, 3 in order and then completes with a CompositeException of both dummy1 and dummy2.

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.

This implements the lawful "monadic bind", the flatMap operation of cats.Monad.

Example

Observable(1, 2, 3).concatMap { x =>
  for {
    _ <- Observable.eval(println(s"Processing $$x"))
    x <- Observable(x, x)
  } yield x
}

Concat vs Merge

The difference between the concat operation and merge is 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. Or in other words concat has deterministic, lawful behavior (being the "monadic bind"), whereas merge has non-deterministic behavior.

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.

Delaying Errors

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.

Example

val dummy1 = new RuntimeException("dummy1")
val dummy2 = new RuntimeException("dummy2")

Observable(1, 2, 3).concatMapDelayErrors {
  case 1 => Observable(1).endWithError(dummy1)
  case 2 => Observable.raiseError(dummy2)
  case x => Observable(x, x)
}

The resulting stream in this example emits 1, 3, 3 in order and then completes with a CompositeException of both dummy1 and dummy2.

Polymorphic version consumeWith that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLift conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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.

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.

Usage:

import scala.concurrent.duration._

(Observable("M", "O", "N", "I", "X") ++ Observable.never)
  .delayOnNext(100.millis)
  .scan("")(_ ++ _)
  .debounce(200.millis)
  .dump("O")

// Output:
// 0: O --> MONIX

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.

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.

Version of delayExecutionWith that can work with generic F[_] tasks, anything that's supported via ObservableLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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.

Suppress duplicate consecutive items emitted by the source.

Example:

// Needed to bring standard Eq instances in scope:
import cats.implicits._

// Yields 1, 2, 1, 3, 2, 4
val stream = Observable(1, 1, 1, 2, 2, 1, 1, 3, 3, 3, 2, 2, 4, 4, 4)
  .distinctUntilChanged

Duplication is detected by using the equality relationship provided by the cats.Eq type class. This allows one to override the equality operation being used (e.g. maybe the default .equals is badly defined, or maybe you want reference equality, so depending on use case).

Cats Eq and Scala Interop

Monix prefers to work with cats.Eq for assessing the equality of elements that have an ordering defined, instead of scala.math.Equiv.

We do this because Scala's Equiv has a default instance defined that's based on universal equality and that's a big problem, because when using the Eq type class, it is universal equality that we want to avoid and there have been countless of bugs in the ecosystem related to both universal equality and Equiv. Thankfully people are working to fix it.

We also do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Defining Eq instance is easy and we can use universal equality in our definitions as well:

import cats.Eq

case class Address(host: String, port: Int)

implicit val eqForAddress: Eq[Address] =
  Eq.fromUniversalEquals

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

Example:

// Needed to bring standard instances in scope:
import cats.implicits._

// Yields 1, 2, 3, 4
val stream = Observable(1, 3, 2, 4, 2, 3, 5, 7, 4)
  .distinctUntilChangedByKey(_ % 2)

Duplication is detected by using the equality relationship provided by the cats.Eq type class. This allows one to override the equality operation being used (e.g. maybe the default .equals is badly defined, or maybe you want reference equality, so depending on use case).

Cats Eq and Scala Interop

Monix prefers to work with cats.Eq for assessing the equality of elements that have an ordering defined, instead of scala.math.Equiv.

We do this because Scala's Equiv has a default instance defined that's based on universal equality and that's a big problem, because when using the Eq type class, it is universal equality that we want to avoid and there have been countless of bugs in the ecosystem related to both universal equality and Equiv. Thankfully people are working to fix it.

We also do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Defining Eq instance is easy and we can use universal equality in our definitions as well:

import cats.Eq

case class Address(host: String, port: Int)

implicit val eqForAddress: Eq[Address] =
  Eq.fromUniversalEquals

Executes the given callback just _after_ the subscription happens.

The executed Task executes after the subscription happens and it will delay the first event being emitted. For example this would delay the emitting of the first event by 1 second:

import monix.eval.Task
import scala.concurrent.duration._

Observable.range(0, 100)
  .doAfterSubscribe(Task.sleep(1.second))

Version of doAfterSubscribe that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

import cats.effect._
import scala.concurrent.duration._
import monix.execution.Scheduler.Implicits.global
import monix.catnap.SchedulerEffect
// Needed for IO.sleep
implicit val timer = SchedulerEffect.timerLiftIO[IO](global)

Observable.range(0, 100)
  .doAfterSubscribeF(IO.sleep(1.second))

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.

import monix.eval.Task

Observable.range(0, 10)
  .doOnComplete(Task(println("Completed!")))

NOTE: in most cases what you want is guaranteeCase or bracketCase. This operator is available for fine-grained control.

Version of doOnComplete that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

import cats.effect.IO

Observable.range(0, 10)
  .doOnCompleteF(IO(println("Completed!")))

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).

Example:

import monix.eval.Task

val stream = Observable.range(0, Int.MaxValue)
  .doOnEarlyStop(Task(println("Stopped early!")))
  .take(100)

NOTE: in most cases what you want is guaranteeCase or bracketCase. This operator is available for fine-grained control.

Version of doOnEarlyStop that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Example:

import cats.effect.IO

val stream = Observable.range(0, Int.MaxValue)
  .doOnEarlyStopF(IO(println("Stopped early!")))
  .take(100)

NOTE: in most cases what you want is guaranteeCase or bracketCase. This operator is available for fine-grained control.

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

Example:

import monix.eval.Task

val dummy = new RuntimeException("dummy")

(Observable.range(0, 10) ++ Observable.raiseError(dummy))
  .doOnError { e =>
    Task(println(s"Triggered error: $$e"))
  }

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.

NOTE: in most cases what you want is guaranteeCase or bracketCase. This operator is available for fine-grained control.

Version of doOnError that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

import cats.effect.IO

val dummy = new RuntimeException("dummy")

(Observable.range(0, 10) ++ Observable.raiseError(dummy))
  .doOnErrorF { e =>
    IO(println(s"Triggered error: $$e"))
  }

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

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.

Version of doOnNextAck that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Version of doOnNext that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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.

For example this observable will have a "delayed execution" of 1 second, plus a delayed first element of another 1 second, therefore it will take a total of 2 seconds for the first element to be emitted:

import monix.eval._
import scala.concurrent.duration._

Observable.range(0, 100)
  .delayExecution(1.second)
  .doOnStart { a =>
    for {
      _ <- Task.sleep(1.second)
      _ <- Task(println(s"Started with: $$a"))
    } yield ()
  }

Version of doOnStart that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

import cats.implicits._
import cats.effect._
import scala.concurrent.duration._
import monix.execution.Scheduler.Implicits.global
import monix.catnap.SchedulerEffect
// Needed for IO.sleep
implicit val timer = SchedulerEffect.timerLiftIO[IO](global)

Observable.range(0, 100)
  .delayExecution(1.second)
  .doOnStartF { a =>
    for {
      _ <- IO.sleep(1.second)
      _ <- IO(println(s"Started with: $$a"))
    } yield ()
  }

Executes the given callback just _before_ the subscription to the source happens.

For example this is equivalent with delayExecution:

import monix.eval.Task
import scala.concurrent.duration._

Observable.range(0, 10)
  .doOnSubscribe(Task.sleep(1.second))

Version of doOnSubscribe that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

For example this is equivalent with delayExecution:

import cats.effect._
import scala.concurrent.duration._
import monix.execution.Scheduler.Implicits.global
import monix.catnap.SchedulerEffect
// Needed for IO.sleep
implicit val timer = SchedulerEffect.timerLiftIO[IO](global)

Observable.range(0, 10)
  .doOnSubscribeF(IO.sleep(1.second))

Executes the given callback when the connection is being cancelled, via the Cancelable reference returned on subscribing to the created observable.

Example:

import monix.eval.Task
import monix.execution.Scheduler

implicit val s = Scheduler.global

val cancelable =
  Observable
    .range(0, Int.MaxValue)
    .doOnSubscriptionCancel(Task(println("Cancelled!")))
    .subscribe()

cancelable.cancel()

NOTE: in most cases what you want is guaranteeCase or bracketCase. This operator is available for fine-grained control.

Version of doOnSubscriptionCancel that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Example:

import cats.effect.IO
import monix.execution.Scheduler

implicit val s = Scheduler.global

val cancelable =
  Observable
    .range(0, Int.MaxValue)
    .doOnSubscriptionCancelF(IO(println("Cancelled!")))
    .subscribe()

cancelable.cancel()

NOTE: in most cases what you want is guaranteeCase or bracketCase. This operator is available for fine-grained control.

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 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.

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.

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.

Emits the given exception instead of onComplete.

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().

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

When an Observable is subscribed with subscribe, it needs a Scheduler, which is going to be injected in the processing pipeline, to be used for managing asynchronous boundaries, scheduling execution with delay, etc.

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 subsumes the effects of subscribeOn, meaning that the subscription logic itself will start on the provided scheduler if forceAsync = true (the default).

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:

import monix.execution.ExecutionModel.AlwaysAsyncExecution

val stream = Observable(1, 2, 3)
  .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.

Only emits those items for which the given predicate holds.

Version of filter that can work with a predicate expressed by a monix.eval.Task.

Version of filterEval that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Only emits those items for which the given predicate doesn't hold.

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.

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.

Alias for concatMap.

NOTE: one primary difference between Monix and other Rx / ReactiveX implementations is that in Monix flatMap is an alias for concatMap and NOT mergeMap.

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.

It's the combination between scan0 and flatMap.

Version of flatScan0 that delays the errors from the emitted streams until the source completes.

Delaying Errors

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.

Version of flatScan that delays the errors from the emitted streams until the source completes.

Delaying Errors

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.

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.

This operation is the "monadic bind", implementing the flatMap operation of cats.Monad.

Concat vs Merge

The difference between the concat operation and merge is 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. Or in other words concat has deterministic, lawful behavior (being the "monadic bind"), whereas merge has non-deterministic behavior.

Alias for concat.

Given evidence that type A has a cats.Monoid implementation, folds the stream with the provided monoid definition.

For streams emitting numbers, this effectively sums them up. For strings, this concatenates them.

Example:

import cats.implicits._

// Yields 10
val stream1 = Observable(1, 2, 3, 4).fold

// Yields "1234"
val stream2 = Observable("1", "2", "3", "4").fold

Note, in case you don't have a Monoid instance in scope, but you feel like you should, try this import:

import cats.instances.all._

Given evidence that type A has a cats.Monoid implementation, folds the stream with the provided monoid definition.

For streams emitting numbers, this effectively sums them up. For strings, this concatenates them.

Example:

import cats.implicits._

// Yields 10
val stream1 = Observable(1, 2, 3, 4).foldL

// Yields "1234"
val stream2 = Observable("1", "2", "3", "4").foldL

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.

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 foldLeft is equivalent to this function being called with an operator always returning true as the first member of its result.

Example:

// Sums first 10 items
val stream1 = Observable.range(0, 1000).foldWhileLeft((0L, 0)) {
  case ((sum, count), e) =>
    val next = (sum + e, count + 1)
    if (count + 1 < 10) Left(next) else Right(next)
}

// Implements exists(predicate)
val stream2 = Observable(1, 2, 3, 4, 5).foldWhileLeft(false) {
  (default, e) =>
    if (e == 3) Right(true) else Left(default)
}

// Implements forall(predicate)
val stream3 = Observable(1, 2, 3, 4, 5).foldWhileLeft(true) {
  (default, e) =>
    if (e != 3) Right(false) else Left(default)
}

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 Left results.

Example:

// Sums first 10 items
val stream1 = Observable.range(0, 1000).foldWhileLeftL((0L, 0)) {
  case ((sum, count), e) =>
    val next = (sum + e, count + 1)
    if (count + 1 < 10) Left(next) else Right(next)
}

// Implements exists(predicate)
val stream2 = Observable(1, 2, 3, 4, 5).foldWhileLeftL(false) {
  (default, e) =>
    if (e == 3) Right(true) else Left(default)
}

// Implements forall(predicate)
val stream3 = Observable(1, 2, 3, 4, 5).foldWhileLeftL(true) {
  (default, e) =>
    if (e != 3) Right(false) else Left(default)
}

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.

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).

Given a routine make sure to execute it whenever the current stream reaches the end, successfully, in error, or canceled.

Implements cats.effect.Bracket.guarantee.

Example:

import monix.eval.Task

Observable.suspend(???).guarantee(Task.eval {
  println("Releasing resources!")
})

Returns a new Observable in which f is scheduled to be executed when the source is completed, in success, error or when cancelled.

Implements cats.effect.Bracket.guaranteeCase.

This would typically be used to ensure that a finalizer will run at the end of the stream.

Example:

import cats.effect.ExitCase
import monix.eval.Task

val stream = Observable.suspend(???).guaranteeCase(err => Task {
  err match {
    case ExitCase.Completed =>
      println("Completed successfully!")
    case ExitCase.Error(e) =>
      e.printStackTrace()
    case ExitCase.Canceled =>
      println("Was stopped early!")
  }
})

NOTE this is using cats.effect.ExitCase to signal the termination condition, like this:

• if completed via onComplete or via Stop signalled by the consumer, then the function receives ExitCase.Completed
• if completed via onError or in certain cases in which errors are detected (e.g. the consumer returns an error), then the function receives ExitCase.Error(e)
• if the subscription was cancelled, then the function receives ExitCase.Canceled

In other words Completed is for normal termination conditions, Error is for exceptions being detected and Canceled is for when the subscription gets canceled.

Version of guaranteeCase that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Version of guarantee that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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.

Creates a new observable from this observable that will emit the start element followed by the upstream elements paired with the separator, and lastly the end element.

Usage sample:

// Yields "begin a : b : c : d end"
Observable("a", "b", "c", "d")
  .intersperse("begin ", " : ", " end")
  .foldLeftL("")(_ ++ _)

Creates a new observable from this observable that will emit a specific separator between every pair of elements.

Usage sample:

// Yields "a : b : c : d"
Observable("a", "b", "c", "d")
  .intersperse(" : ")
  .foldLeftL("")(_ ++ _)

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.

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 element of the applied function.

Similar to scan, but the supplied function returns a tuple of the next accumulator state and the result type emitted by the returned observable.

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

Example:

import monix.eval.Task
import scala.concurrent.duration._

Observable.range(0, 100)
  .mapEval(x => Task(x).delayExecution(1.second))

Version of mapEval that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

Example:

import cats.implicits._
import cats.effect.IO
import scala.concurrent.duration._
import monix.execution.Scheduler.Implicits.global
import monix.catnap.SchedulerEffect
// Needed for IO.sleep
implicit val timer = SchedulerEffect.timerLiftIO[IO](global)

Observable.range(0, 100).mapEvalF { x =>
  IO.sleep(1.second) *> IO(x)
}

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 returning them preserving original order.

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 mapEval.

Version of mapParallelOrderedF that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ...

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 mapEval.

Version of mapParallelUnordered that can work with generic F[_] tasks, anything that's supported via monix.eval.TaskLike conversions.

So you can work among others with:

• cats.effect.IO
• monix.eval.Coeval
• scala.concurrent.Future
• ... Note that when the specified parallelism is 1, it has the same behavior as mapEval.

Given a cats.Order over the stream's elements, returns the maximum element in the stream.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

// Yields Observable(20)
val stream1 = Observable(10, 7, 6, 8, 20, 3, 5).max

// Yields Observable.empty
val stream2 = Observable.empty[Int].max

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

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.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

case class Person(name: String, age: Int)

// Yields Observable(Person("Alex", 34))
Observable(Person("Alex", 34), Person("Alice", 27))
  .maxBy(_.age)

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

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.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

case class Person(name: String, age: Int)

// Yields Some(Person("Alex", 34))
Observable(Person("Alex", 34), Person("Alice", 27))
  .maxByL(_.age)

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

Given a cats.Order over the stream's elements, returns the maximum element in the stream.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

// Yields Some(20)
val stream1 = Observable(10, 7, 6, 8, 20, 3, 5).maxL

// Yields Observable.empty
val stream2 = Observable.empty[Int].maxL

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

Concurrently merges the observables emitted by the source, into a single observable.

Equivalence with mergeMap

The merge operation is mergeMap with the identity function:

stream.merge <-> stream.mergeMap(x => x)

Concat vs Merge

The difference between the concat operation and merge is 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. Or in other words concat has deterministic, lawful behavior (being the "monadic bind"), whereas merge has non-deterministic behavior.

Visual Example

streamA: a1 -- -- a2 -- -- a3 -- a4 -- --
streamB: b1 -- -- b2 -- b3 -- -- -- -- b4

result: a1, b1, a2, b2, b3, a3, a4, b4

Delaying Errors

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.

Concurrently merges the observables emitted by the source with the given generator function into a single observable.

Concat vs Merge

The difference between the concat operation and merge is 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. Or in other words concat has deterministic, lawful behavior (being the "monadic bind"), whereas merge has non-deterministic behavior.

Example

Observable(1, 2, 3).mergeMap { x =>
  Observable.eval(println(s"Processing $$x"))
    .executeAsync
    .flatMap(_ => Observable(x, x))
}

In this example the source will yield 3 streams and those 3 streams are being subscribed immediately, therefore the order of the events will be non-deterministic, as the streams will be evaluated concurrently.

Visual Example

streamA: a1 -- -- a2 -- -- a3 -- a4 -- --
streamB: b1 -- -- b2 -- b3 -- -- -- -- b4

result: a1, b1, a2, b2, b3, a3, a4, b4

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.

Concat vs Merge

The difference between the concat operation and merge is 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. Or in other words concat has deterministic, lawful behavior (being the "monadic bind"), whereas merge has non-deterministic behavior.

Delaying Errors

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.

Given a cats.Order over the stream's elements, returns the minimum element in the stream.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

// Yields Observable(3)
val stream1 =
  Observable(10, 7, 6, 8, 20, 3, 5).min

// Yields Observable.empty
val stream2 =
  Observable.empty[Int].min

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

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.

Example:

// Needed to bring the standard Order instances in scope:
import cats.implicits._

case class Person(name: String, age: Int)

// Yields Observable(Person("Alice", 27))
val stream = Observable(Person("Alex", 34), Person("Alice", 27))
  .minBy(_.age)

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

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.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

case class Person(name: String, age: Int)

// Yields Some(Person("Alice", 27))
Observable(Person("Alex", 34), Person("Alice", 27))
  .minByL(_.age)

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

Given a cats.Order over the stream's elements, returns the minimum element in the stream.

Example

// Needed to bring the standard Order instances in scope:
import cats.implicits._

// Yields Some(3)
val stream1 =
  Observable(10, 7, 6, 8, 20, 3, 5).minL

// Yields None
val stream2 =
  Observable.empty[Int].minL

Cats Order and Scala Interop

Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

Building a cats.Order is easy to do if you already have a Scala Ordering instance:

import cats.Order

case class Person(name: String, age: Int)

// Starting from a Scala Ordering
implicit val scalaOrderingForPerson: Ordering[Person] =
  new Ordering[Person] {
    def compare(x: Person, y: Person): Int =
      x.age.compareTo(y.age) match {
        case 0 => x.name.compareTo(y.name)
        case o => o
      }
  }

// Building a cats.Order from it
implicit val catsOrderForPerson: Order[Person] =
  Order.fromOrdering

You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

val scalaOrdering = catsOrderForPerson.toOrdering

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).

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

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
import monix.execution.Scheduler.Implicits.global
val io = Scheduler.io("my-io")

Observable(1, 2, 3).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:

import monix.reactive.OverflowStrategy.Unbounded

Observable.suspend(???)
  .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.

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.

This operators takes a possibly pure Observable, transforms it to Hot Observable in the scope of supplied function and then returns a pure Observable again. The function allows specyfing underlying monix.reactive.subjects.Subject by means of monix.reactive.Pipe.

Example

import monix.reactive._
import monix.eval.Task
import scala.concurrent.duration._

val obs = Observable(1, 2, 3)
  .doOnNext(i => Task(println(s"Produced $$i")).delayExecution(1.second))

def consume(name: String, obs: Observable[Int]): Observable[Unit] =
  obs.mapEval(i => Task(println(s"$$name: got $$i")))

obs.pipeThroughSelector(Pipe.replay[Int], { hot: Observable[Int] =>
  Observable(
    consume("Consumer 1", hot),
    consume("Consumer 2", hot).delayExecution(2.second)
  ).merge
})

Output

Produced 1 Consumer 1: got 1 Consumer 2: got 1 Produced 2 Consumer 1: got 2 Consumer 2: got 2 Produced 3 Consumer 1: got 3 Consumer 2: got 3

Note how Consumer 2 received the same amount of elements as Consumer 1 despite subscribing later because of underlying 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 PublishSubject.

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

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.

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

For FP code these functions shouldn't be called until "the end of the world", which is to say at the end of the program (for a console app), or at the end of a web request.

Otherwise for modifying or operating on streams, prefer its pure functions like publishSelector for sharing the data source, or map or flatMap for operating on its events. Or in case of specialized logic, prefer to suspend these side effects via Observable.suspend. Monix also provides Task which can also be used for suspending side effects and the Task was built to interop well with Observable.

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

This operators takes a possibly pure Observable, transforms it to Hot Observable in the scope of supplied function and then returns a pure Observable again.

Example

import monix.reactive._
import monix.eval.Task
import scala.concurrent.duration._

val obs = Observable(1, 2, 3)
  .doOnNext(i => Task(println(s"Produced $$i")).delayExecution(1.second))

def consume(name: String, obs: Observable[Int]): Observable[Unit] =
  obs.mapEval(i => Task(println(s"$$name: got $$i")))

obs.publishSelector { hot =>
  Observable(
    consume("Consumer 1", hot),
    consume("Consumer 2", hot).delayExecution(2.second)
  ).merge
}

Output

Produced 1 Consumer 1: got 1 Produced 2 Consumer 1: got 2 Consumer 2: got 2 Produced 3 Consumer 1: got 3 Consumer 2: got 3

Note how Consumer 2 received less elements because it subscribed later.