Packages

abstract class Observable[+A] extends Serializable

The Observable type that implements the Reactive Pattern.

Provides methods of subscribing to the Observable and operators for combining observable sources, filtering, modifying, throttling, buffering, error handling and others.

See the available documentation at: https://monix.io

Self Type
Observable[A]
Source
Observable.scala
Linear Supertypes
Ordering
  1. Alphabetic
  2. By Inheritance
Inherited
  1. Observable
  2. Serializable
  3. AnyRef
  4. Any
  1. Hide All
  2. Show All
Visibility
  1. Public
  2. Protected

Instance Constructors

  1. new Observable()

Abstract Value Members

  1. abstract def unsafeSubscribeFn(subscriber: Subscriber[A]): Cancelable

    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.

    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.

    Annotations
    @UnsafeProtocol() @UnsafeBecauseImpure()

Concrete Value Members

  1. final def !=(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  2. final def ##: Int
    Definition Classes
    AnyRef → Any
  3. final def ++[B >: A](other: => Observable[B]): Observable[B]

    Concatenates the source with another 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
    

  4. final def +:[B >: A](elem: B): Observable[B]

    Alias for prepend.

  5. final def :+[B >: A](elem: B): Observable[B]

    Alias for append.

  6. final def ==(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  7. final def ambWith[B >: A](other: Observable[B]): Observable[B]

    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.

  8. final def append[B >: A](elem: B): Observable[B]

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

  9. final def appendAll[B >: A](other: Observable[B]): Observable[B]

    A strict variant of ++.

  10. final def asInstanceOf[T0]: T0
    Definition Classes
    Any
  11. final def asyncBoundary[B >: A](overflowStrategy: OverflowStrategy[B]): Observable[B]

    Forces a buffered asynchronous boundary.

    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
    overflowStrategy

    - the overflow strategy used for buffering, which specifies what to do in case we're dealing with a slow consumer - should an unbounded buffer be used, should back-pressure be applied, should the pipeline drop newer or older events, should it drop the whole buffer? See OverflowStrategy for more details.

  12. final def behavior[B >: A](initialValue: B)(implicit s: Scheduler): ConnectableObservable[B]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

    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.

    Annotations
    @UnsafeBecauseImpure()
  13. final def bracket[B](use: (A) => Observable[B])(release: (A) => Task[Unit]): Observable[B]

    Implementation of bracket from cats.effect.Bracket.

    Implementation of bracket from cats.effect.Bracket.

    See documentation.

  14. final def bracketCase[B](use: (A) => Observable[B])(release: (A, ExitCase[Throwable]) => Task[Unit]): Observable[B]

    Implementation of bracketCase from cats.effect.Bracket.

    Implementation of bracketCase from cats.effect.Bracket.

    See documentation.

  15. final def bracketCaseF[F[_], B](use: (A) => Observable[B])(release: (A, ExitCase[Throwable]) => F[Unit])(implicit F: TaskLike[F]): Observable[B]

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

    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
    • ...
  16. final def bracketF[F[_], B](use: (A) => Observable[B])(release: (A) => F[Unit])(implicit F: TaskLike[F]): Observable[B]

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

    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
    • ...
  17. final def bufferIntrospective(maxSize: Int): Observable[List[A]]

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

    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)
  18. final def bufferSliding(count: Int, skip: Int): Observable[Seq[A]]

    Returns an observable that emits buffers of items it collects from the source observable.

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

    the maximum size of each buffer before it should be emitted

    skip

    how many items emitted by the source observable should be skipped before starting a new buffer. Note that when skip and count are equal, this is the same operation as bufferTumbling(count)

  19. final def bufferTimed(timespan: FiniteDuration): Observable[Seq[A]]

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

    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.

    timespan

    the interval of time at which it should emit the buffered bundle

  20. final def bufferTimedAndCounted(timespan: FiniteDuration, maxCount: Int): Observable[Seq[A]]

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

    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.

    timespan

    the interval of time at which it should emit the buffered bundle

    maxCount

    is the maximum bundle size, after which the buffered bundle gets forcefully emitted

  21. final def bufferTimedWithPressure[AA >: A](period: FiniteDuration, maxSize: Int, sizeOf: (AA) => Int = (_: AA) => 1): Observable[Seq[AA]]

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

    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.

    period

    the interval of time at which it should emit the buffered bundle

    maxSize

    is the maximum buffer size, after which the source starts being back-pressured

    sizeOf

    is the function to compute the weight of each element in the buffer

  22. final def bufferTumbling(count: Int): Observable[Seq[A]]

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

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

    the maximum size of each buffer before it should be emitted

  23. final def bufferWhile(p: (A) => Boolean): Observable[Seq[A]]

    Buffers elements while predicate returns true, after which it emits the buffered events as a single bundle and creates a new buffer.

    Buffers elements while predicate returns true, after which it emits the buffered events as a single bundle and creates a new buffer.

    Usage:

    import monix.eval.Task
    
    Observable(1, 1, 1, 2, 2, 1, 3)
      .bufferWhile(_ == 1)
      .doOnNext(l => Task(println(s"Emitted batch $$l")))
    
    // Emitted batch List(1, 1, 1)
    // Emitted batch List(2)
    // Emitted batch List(2, 1)
    // Emitted batch List(3)
    See also

    bufferWhileInclusive for a similar operator that includes the value that caused predicate to return false

  24. final def bufferWhileInclusive(p: (A) => Boolean): Observable[Seq[A]]

    Buffers elements while predicate returns true, after which it emits the buffered events as a single bundle, including the value that caused predicate to return false and creates a new buffer.

    Buffers elements while predicate returns true, after which it emits the buffered events as a single bundle, including the value that caused predicate to return false and creates a new buffer.

    Usage:

    import monix.eval.Task
    
    Observable(1, 1, 1, 2, 2, 1, 3)
      .bufferWhileInclusive(_ == 1)
      .doOnNext(l => Task(println(s"Emitted batch $$l")))
    
    // Emitted batch List(1, 1, 1, 2)
    // Emitted batch List(2)
    // Emitted batch List(1, 3)
    See also

    bufferWhile for a similar operator that does not include the value that caused predicate to return false

  25. final def bufferWithSelector[S](selector: Observable[S], maxSize: Int): Observable[Seq[A]]

    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.

    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.

    selector

    is the observable that triggers the signaling of the current buffer

    maxSize

    is the maximum bundle size, after which the source starts being back-pressured

  26. final def bufferWithSelector[S](selector: Observable[S]): Observable[Seq[A]]

    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.

    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.

    selector

    is the observable that triggers the signaling of the current buffer

  27. final def cache(maxCapacity: Int): Observable[A]

    Caches the emissions from the source Observable and replays them in order to any subsequent Subscribers.

    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.

    maxCapacity

    is the maximum buffer size after which old events start being dropped (according to what happens when using ReplaySubject.createLimited)

    returns

    an Observable that, when first subscribed to, caches all of its items and notifications for the benefit of subsequent subscribers

    Annotations
    @UnsafeBecauseImpure()
  28. final def cache: Observable[A]

    Caches the emissions from the source Observable and replays them in order to any subsequent Subscribers.

    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.

    returns

    an Observable that, when first subscribed to, caches all of its items and notifications for the benefit of subsequent subscribers

    Annotations
    @UnsafeBecauseImpure()
  29. def clone(): AnyRef
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.CloneNotSupportedException]) @native() @HotSpotIntrinsicCandidate()
  30. final def collect[B](pf: PartialFunction[A, B]): Observable[B]

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

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

    pf

    the function that filters and maps the source

    returns

    an observable that emits the transformed items by the given partial function

  31. final def collectWhile[B](pf: PartialFunction[A, B]): Observable[B]

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

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

    pf

    the function that filters and maps the source

    returns

    an observable that emits the transformed items by the given partial function until it is contained in the function's domain

  32. final def combineLatest[B](other: Observable[B]): Observable[(A, B)]

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

    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.

    other

    is an observable that gets paired with the source

  33. final def combineLatestMap[B, R](other: Observable[B])(f: (A, B) => R): Observable[R]

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

    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.

    other

    is an observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  34. final def completed: Observable[Nothing]

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

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

    returns

    an empty Observable that only calls onCompleted or onError, based on which one is called by the source Observable

  35. final def completedF[F[_]](implicit F: TaskLift[F]): F[Unit]

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

    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
    • ...
  36. final def completedL: Task[Unit]

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

  37. final def concat[B](implicit ev: <:<[A, Observable[B]]): Observable[B]

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

    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
    

    returns

    an observable that emits the merged events of all streams created by the source

  38. final def concatDelayErrors[B](implicit ev: <:<[A, Observable[B]]): Observable[B]

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

    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.

    returns

    an observable that emits the merged events of all streams created by the source

  39. final def concatMap[B](f: (A) => Observable[B]): Observable[B]

    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.

    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.

    f

    is a generator for the streams being concatenated

    returns

    an observable that emits the merged events of all streams created by the source

  40. final def concatMapDelayErrors[B](f: (A) => Observable[B]): Observable[B]

    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.

    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.

    f

    is a generator for the streams being concatenated

    returns

    an observable that emits the merged events of all streams created by the source

  41. final def concatMapIterable[B](f: (A) => Iterable[B]): Observable[B]

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

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

    Example

    Observable(1, 2, 3).concatMapIterable( x => List(x, x * 10, x * 100))

    Visual Example

    stream: 1 -- -- 2 -- -- 3 -- --
    result: 1, 10, 100, 2, 20, 200, 3, 30, 300
    

    f

    is a generator for the sequences being concatenated

  42. final def consumeWith[R](f: Consumer[A, R]): Task[R]

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

  43. final def consumeWithF[F[_], R](f: Consumer[A, R])(implicit F: TaskLift[F]): F[R]

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

    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
    • ...
  44. final def count: Observable[Long]

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

    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.

  45. final def countL: Task[Long]

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

  46. final def debounce(timeout: FiniteDuration): Observable[A]

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

    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
    timeout

    the length of the window of time that must pass after the emission of an item from the source observable in which that observable emits no items in order for the item to be emitted by the resulting observable

    See also

    echoOnce for a similar operator that also mirrors the source observable

  47. final def debounceRepeated(period: FiniteDuration): Observable[A]

    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.

    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.

    period

    the length of the window of time that must pass after the emission of an item from the source Observable in which that Observable emits no items in order for the item to be emitted by the resulting Observable at regular intervals, also determined by period

    See also

    echoRepeated for a similar operator that also mirrors the source observable

  48. final def debounceTo[B](timeout: FiniteDuration, f: (A) => Observable[B]): Observable[B]

    Doesn't emit anything until a timeout period passes without the source emitting anything.

    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.

    timeout

    the length of the window of time that must pass after the emission of an item from the source Observable in which that Observable emits no items in order for the item to be emitted by the resulting Observable

    f

    is a function that receives the last element generated by the source, generating an observable to be subscribed when the source is timing out

  49. final def defaultIfEmpty[B >: A](default: => B): Observable[B]

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

  50. final def delayExecution(timespan: FiniteDuration): Observable[A]

    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 for a specified amount of time before passing it on to the source Observable.

    timespan

    is the time to wait before the subscription is being initiated.

  51. final def delayExecutionWith[B](trigger: Observable[B]): Observable[A]

    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.

    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.

    trigger

    the observable that must either emit an item or complete in order for the source to be subscribed.

  52. final def delayExecutionWithF[F[_], B](trigger: F[B])(implicit F: ObservableLike[F]): Observable[A]

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

    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
    • ...
  53. final def delayOnComplete(delay: FiniteDuration): Observable[A]

    Delays emitting the final onComplete event by the specified amount.

  54. final def delayOnNext(duration: FiniteDuration): Observable[A]

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

    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.

    duration

    - the delay to shift the source by

    returns

    the source Observable shifted in time by the specified delay

  55. final def delayOnNextBySelector[B](selector: (A) => Observable[B]): Observable[A]

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

    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.

    selector

    is a function that returns an Observable for each item emitted by the source Observable, which is then used to delay the emission of that item by the resulting Observable until the Observable returned from selector emits an item

    returns

    the source Observable shifted in time by the specified delay

  56. final def dematerialize[B](implicit ev: <:<[A, Notification[B]]): Observable[B]

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

  57. final def distinctUntilChanged[AA >: A](implicit A: Eq[AA]): Observable[AA]

    Suppress duplicate consecutive items emitted by the source.

    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
    A

    is the cats.Eq instance that defines equality for the elements emitted by the source

  58. final def distinctUntilChangedByKey[K](key: (A) => K)(implicit K: Eq[K]): Observable[A]

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

    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
    key

    is a function that returns a K key for each element, a value that's then used to do the deduplication

    K

    is the cats.Eq instance that defines equality for the key type K

  59. final def doAfterSubscribe(task: Task[Unit]): Observable[A]

    Executes the given callback just _after_ the subscription happens.

    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))
    See also

    doOnSubscribe for executing a callback just before a subscription happens.

  60. final def doAfterSubscribeF[F[_]](task: F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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 cats.effect.Timer
    import scala.concurrent.duration._
    import monix.execution.Scheduler.Implicits.global
    import monix.catnap.SchedulerEffect
    // Needed for IO.sleep
    implicit val timer: Timer[IO] = SchedulerEffect.timerLiftIO[IO](global)
    
    Observable.range(0, 100)
      .doAfterSubscribeF(IO.sleep(1.second))
  61. final def doOnComplete(task: Task[Unit]): Observable[A]

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

    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.

    task

    the task to execute when the onComplete event gets emitted

  62. final def doOnCompleteF[F[_]](task: F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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!")))
  63. final def doOnEarlyStop(task: Task[Unit]): Observable[A]

    Executes the given task 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).

    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.

  64. final def doOnEarlyStopF[F[_]](task: F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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.

  65. final def doOnError(cb: (Throwable) => Task[Unit]): Observable[A]

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

    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.

  66. final def doOnErrorF[F[_]](cb: (Throwable) => F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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"))
      }
  67. final def doOnNext(cb: (A) => Task[Unit]): Observable[A]

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

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

    returns

    a new Observable that executes the specified callback for each element

    See also

    doOnNext for a simpler version that doesn't allow asynchronous execution.

  68. final def doOnNextAck(cb: (A, Ack) => Task[Unit]): Observable[A]

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

    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.

    See also

    doOnNextAckF for a version that can do evaluation with any data type via monix.eval.TaskLike

  69. final def doOnNextAckF[F[_]](cb: (A, Ack) => F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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
    • ...
  70. final def doOnNextF[F[_]](cb: (A) => F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

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

    a new Observable that executes the specified callback for each element

  71. final def doOnStart(cb: (A) => Task[Unit]): Observable[A]

    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 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 ()
      }
    returns

    a new Observable that executes the specified task only for the first element

  72. final def doOnStartF[F[_]](cb: (A) => F[Unit])(implicit F: Effect[F]): Observable[A]

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

    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 cats.effect.Timer
     import scala.concurrent.duration._
     import monix.execution.Scheduler.Implicits.global
     import monix.catnap.SchedulerEffect
     // Needed for IO.sleep
    implicit val timer: Timer[IO] = 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 ()
       }
  73. final def doOnSubscribe(task: Task[Unit]): Observable[A]

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

    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))
    See also

    doAfterSubscribe for executing a callback just after a subscription happens.

  74. final def doOnSubscribeF[F[_]](task: F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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 cats.effect.Timer
     import scala.concurrent.duration._
     import monix.execution.Scheduler.Implicits.global
     import monix.catnap.SchedulerEffect
     // Needed for IO.sleep
    implicit val timer: Timer[IO] = SchedulerEffect.timerLiftIO[IO](global)
    
     Observable.range(0, 10)
       .doOnSubscribeF(IO.sleep(1.second))
  75. final def doOnSubscriptionCancel(task: Task[Unit]): Observable[A]

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

    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.

  76. final def doOnSubscriptionCancelF[F[_]](task: F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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.

  77. final def drop(n: Long): Observable[A]

    Drops the first n elements (from the start).

    Drops the first n elements (from the start).

    n

    the number (Long) of elements to drop

    returns

    a new Observable that drops the first n elements emitted by the source

  78. final def drop(n: Int): Observable[A]

    Overload of drop(Long).

  79. final def dropByTimespan(timespan: FiniteDuration): Observable[A]

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

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

    timespan

    the window of time during which the new observable must drop events emitted by the source

  80. final def dropLast(n: Int): Observable[A]

    Drops the last n elements (from the end).

    Drops the last n elements (from the end).

    n

    the number of elements to drop

    returns

    a new Observable that drops the first n elements emitted by the source

  81. final def dropUntil(trigger: Observable[Any]): Observable[A]

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

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

    trigger

    the observable that has to emit an item before the source begin to be mirrored by the resulting observable

  82. final def dropWhile(p: (A) => Boolean): Observable[A]

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

  83. final def dropWhileInclusive(p: (A) => Boolean): Observable[A]

    Drops the longest prefix of elements that satisfy the given predicate, inclusive of the value that caused predicate to return false and returns a new observable that emits the rest.

  84. final def dropWhileWithIndex(p: (A, Int) => Boolean): Observable[A]

    Drops the longest prefix of elements that satisfy the given function 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.

  85. final def dump(prefix: String, out: PrintStream = System.out): Observable[A]

    Utility that can be used for debugging purposes.

  86. final def echoOnce(timeout: FiniteDuration): Observable[A]

    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.

    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.

    timeout

    the window of silence that must pass in order for the observable to echo the last item

  87. final def echoRepeated(timeout: FiniteDuration): Observable[A]

    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.

    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.

    timeout

    the window of silence that must pass in order for the observable to start echoing the last item

  88. final def endWith[B >: A](elems: Seq[B]): Observable[B]

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

  89. final def endWithError(error: Throwable): Observable[A]

    Emits the given exception instead of onComplete.

    Emits the given exception instead of onComplete.

    error

    the exception to emit onComplete

    returns

    a new Observable that emits an exception onComplete

  90. final def eq(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  91. def equals(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef → Any
  92. final def executeAsync: Observable[A]

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

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

  93. final def executeOn(s: Scheduler, forceAsync: Boolean = true): Observable[A]

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

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

    s

    is the Scheduler to use for overriding the default scheduler and for forcing an asynchronous boundary if forceAsync is true

    forceAsync

    indicates whether an asynchronous boundary should be forced right before the subscription of the source Observable, managed by the provided Scheduler

    returns

    a new Observable that mirrors the source on subscription, but that uses the provided scheduler for overriding the default and possibly force an extra asynchronous boundary on execution

    See also

    observeOn and subscribeOn.

  94. final def executeWithModel(em: ExecutionModel): Observable[A]

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

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

    is the ExecutionModel that will be used when evaluating the source.

  95. final def exists(p: (A) => Boolean): Observable[Boolean]

    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 an Observable which emits a single value, either true, in case the given predicate holds for at least one item, or false otherwise.

    p

    is a function that evaluates the items emitted by the source Observable, returning true if they pass the filter

    returns

    an Observable that emits only true or false in case the given predicate holds or not for at least one item

  96. final def existsL(p: (A) => Boolean): Task[Boolean]

    Returns a Task which emits 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.

    p

    is a function that evaluates the items emitted by the source, returning true if they pass the filter

    returns

    a task that emits true or false in case the given predicate holds or not for at least one item

  97. final def failed: Observable[Throwable]

    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.

  98. final def filter(p: (A) => Boolean): Observable[A]

    Only emits those items for which the given predicate holds.

    Only emits those items for which the given predicate holds.

    p

    a function that evaluates the items emitted by the source returning true if they pass the filter

    returns

    a new observable that emits only those items in the source for which the filter evaluates as true

    See also

    filterEval for a version that works with a monix.eval.Task.

    filterEvalF for a version that works with a generic F[_] (e.g. cats.effect.IO, Scala's Future), powered by monix.eval.TaskLike

  99. final def filterEval(p: (A) => Task[Boolean]): Observable[A]

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

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

    See also

    filterEvalF for a version that works with a generic F[_] (e.g. cats.effect.IO, Scala's Future), powered by monix.eval.TaskLike

  100. final def filterEvalF[F[_]](p: (A) => F[Boolean])(implicit F: TaskLike[F]): Observable[A]

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

    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
    • ...
  101. final def filterNot(p: (A) => Boolean): Observable[A]

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

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

    p

    a function that evaluates the items emitted by the source returning true if they should be filtered out

    returns

    a new observable that emits only those items in the source for which the filter evaluates as false

  102. final def find(p: (A) => Boolean): Observable[A]

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

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

    p

    is a function that evaluates the items emitted by the source Observable, returning true if they pass the filter

    returns

    an Observable that emits only the first item in the original Observable for which the filter evaluates as true

  103. final def findL(p: (A) => Boolean): Task[Option[A]]

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

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

    p

    is a function that evaluates the items emitted by the source observable, returning true if they pass the filter

    returns

    a task that emits the first item in the source observable for which the filter evaluates as true

  104. final def firstL: Task[A]

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

    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.

  105. final def firstOptionL: Task[Option[A]]

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

    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.

  106. final def firstOrElse[B >: A](default: => B): Observable[B]

    Alias for headOrElse.

  107. final def firstOrElseL[B >: A](default: => B): Task[B]

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

    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.

  108. final def flatMap[B](f: (A) => Observable[B]): Observable[B]

    Alias for concatMap.

    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.

  109. final def flatMapDelayErrors[B](f: (A) => Observable[B]): Observable[B]

    Alias of concatMapDelayErrors.

  110. final def flatMapIterable[B](f: (A) => Iterable[B]): Observable[B]

    Alias for concatMapIterable

    Alias for concatMapIterable

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

  111. final def flatMapLatest[B](f: (A) => Observable[B]): Observable[B]

    Alias of switchMap.

  112. final def flatScan[R](seed: => R)(op: (R, A) => Observable[R]): Observable[R]

    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.

    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.

    See also

    flatScan0 for the version that emits seed element at the beginning

  113. final def flatScan0[R](seed: => R)(op: (R, A) => Observable[R]): Observable[R]

    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.

    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.

  114. final def flatScan0DelayErrors[R](seed: => R)(op: (R, A) => Observable[R]): Observable[R]

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

    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.

    See also

    flatScan0

  115. final def flatScanDelayErrors[R](seed: => R)(op: (R, A) => Observable[R]): Observable[R]

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

    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.

    See also

    flatScan

  116. final def flatten[B](implicit ev: <:<[A, Observable[B]]): Observable[B]

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

    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.

    returns

    an observable that emits the merged events of all streams created by the source

  117. final def flattenDelayErrors[B](implicit ev: <:<[A, Observable[B]]): Observable[B]

    Alias for concatDelayErrors.

  118. final def flattenLatest[B](implicit ev: <:<[A, Observable[B]]): Observable[B]

    Alias for switch.

  119. final def fold[AA >: A](implicit A: Monoid[AA]): Observable[AA]

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

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

    is the cats.Monoid type class instance that's needed in scope for folding the source

    returns

    the result of combining all elements of the source, or the defined Monoid.empty element in case the stream is empty

    See also

    foldL for the version that returns a task instead of an observable.

  120. final def foldL[AA >: A](implicit A: Monoid[AA]): Task[AA]

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

    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
    A

    is the cats.Monoid type class instance that's needed in scope for folding the source

    returns

    the result of combining all elements of the source, or the defined Monoid.empty element in case the stream is empty

    See also

    fold for the version that returns an observable instead of a task.

  121. final def foldLeft[R](seed: => R)(op: (R, A) => R): Observable[R]

    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 this Observable, going left to right and returns a new Observable that emits only one item before onComplete.

    seed

    is the initial state, specified as a possibly lazy value; it gets evaluated when the subscription happens and if it triggers an error then the subscriber will get immediately terminated with an error

    op

    is an operator that will fold the signals of the source observable, returning the next state

  122. final def foldLeftL[R](seed: => R)(op: (R, A) => R): Task[R]

    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.

  123. final def foldWhileLeft[S](seed: => S)(op: (S, A) => Either[S, S]): Observable[S]

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

    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)
    }
    seed

    is the initial state, specified as a possibly lazy value; it gets evaluated when the subscription happens and if it triggers an error then the subscriber will get immediately terminated with an error

    op

    is the binary operator returning either Left, signaling that the state should be evolved or a Right, signaling that the process can be short-circuited and the result returned immediately

    returns

    the result of inserting op between consecutive elements of this observable, going from left to right with the seed as the start value, or seed if the observable is empty

    See also

    foldWhileLeftL for a version that returns a task instead of an observable.

  124. final def foldWhileLeftL[S](seed: => S)(op: (S, A) => Either[S, S]): Task[S]

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

    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)
    }
    seed

    is the initial state, specified as a possibly lazy value; it gets evaluated when the subscription happens and if it triggers an error then the subscriber will get immediately terminated with an error

    op

    is the binary operator returning either Left, signaling that the state should be evolved or a Right, signaling that the process can be short-circuited and the result returned immediately

    returns

    the result of inserting op between consecutive elements of this observable, going from left to right with the seed as the start value, or seed if the observable is empty

    See also

    foldWhileLeft for a version that returns an observable instead of a task.

  125. final def forall(p: (A) => Boolean): Observable[Boolean]

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

    p

    is a function that evaluates the items emitted by the source Observable, returning true if they pass the filter

    returns

    an Observable that emits only true or false in case the given predicate holds or not for all the items

  126. final def forallL(p: (A) => Boolean): Task[Boolean]

    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.

    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.

    p

    is a function that evaluates the items emitted by the source observable, returning true if they pass the filter

    returns

    a task that emits only true or false in case the given predicate holds or not for all the items

  127. final def foreach(cb: (A) => Unit)(implicit s: Scheduler): CancelableFuture[Unit]

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

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

    Annotations
    @UnsafeBecauseImpure()
  128. final def foreachL(cb: (A) => Unit): Task[Unit]

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

  129. final def getClass(): Class[_ <: AnyRef]
    Definition Classes
    AnyRef → Any
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  130. final def groupBy[K](keySelector: (A) => K)(implicit os: Synchronous[Nothing] = OverflowStrategy.Unbounded): Observable[GroupedObservable[K, A]]

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

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

    keySelector

    a function that extracts the key for each item

  131. final def guarantee(f: Task[Unit]): Observable[A]

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

    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!")
    })
    f

    is the function to execute on early stop

  132. final def guaranteeCase(f: (ExitCase[Throwable]) => Task[Unit]): Observable[A]

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

    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.

    f

    is the finalizer to execute when streaming is terminated, by successful completion, error or cancellation; for specifying the side effects to use

  133. final def guaranteeCaseF[F[_]](f: (ExitCase[Throwable]) => F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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
    • ...
  134. final def guaranteeF[F[_]](f: F[Unit])(implicit F: TaskLike[F]): Observable[A]

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

    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
    • ...
  135. def hashCode(): Int
    Definition Classes
    AnyRef → Any
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  136. final def head: Observable[A]

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

  137. final def headL: Task[A]

    Alias for firstL.

  138. final def headOptionL: Task[Option[A]]

    Alias for firstOptionL.

  139. final def headOrElse[B >: A](default: => B): Observable[B]

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

  140. final def headOrElseL[B >: A](default: => B): Task[B]

    Alias for firstOrElseL.

  141. final def ignoreElements: Observable[Nothing]

    Alias for completed.

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

    returns

    an empty sequence that only calls onCompleted or onError, based on which one is called by the source Observable

  142. final def interleave[B >: A](other: Observable[B]): Observable[B]

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

    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.

    other

    is an observable that interleaves with the source

    returns

    a new observable sequence that alternates emission of the items from both child streams

  143. final def intersperse[B >: A](start: B, separator: B, end: B): Observable[B]

    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.

    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("")(_ ++ _)
    start

    is the first element emitted

    separator

    is the separator

    end

    the last element emitted

  144. final def intersperse[B >: A](separator: B): Observable[B]

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

    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("")(_ ++ _)
    separator

    is the separator

  145. final def isEmpty: Observable[Boolean]

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

  146. final def isEmptyL: Task[Boolean]

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

  147. final def isInstanceOf[T0]: Boolean
    Definition Classes
    Any
  148. final def last: Observable[A]

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

  149. final def lastL: Task[A]

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

    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.

  150. final def lastOptionL: Task[Option[A]]

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

    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.

  151. final def lastOrElseL[B >: A](default: => B): Task[B]

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

    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.

  152. final def liftByOperator[B](operator: Operator[A, B]): Observable[B]

    Transforms the source using the given operator.

  153. final def map[B](f: (A) => B): Observable[B]

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

  154. final def mapAccumulate[S, R](seed: => S)(op: (S, A) => (S, R)): Observable[R]

    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.

    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.

  155. final def mapEval[B](f: (A) => Task[B]): Observable[B]

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

    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))
    See also

    mapEvalF for a version that works with a generic F[_] (e.g. cats.effect.IO, Scala's Future), powered by monix.eval.TaskLike

  156. final def mapEvalF[F[_], B](f: (A) => F[B])(implicit F: TaskLike[F]): Observable[B]

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

    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 cats.effect.Timer
    import scala.concurrent.duration._
    import monix.execution.Scheduler.Implicits.global
    import monix.catnap.SchedulerEffect
    // Needed for IO.sleep
    implicit val timer: Timer[IO] = SchedulerEffect.timerLiftIO[IO](global)
    
    Observable.range(0, 100).mapEvalF { x =>
      IO.sleep(1.second) *> IO(x)
    }
    See also

    mapEval for a version specialized for Task

  157. final def mapParallelOrdered[B](parallelism: Int)(f: (A) => Task[B])(implicit os: OverflowStrategy[B] = OverflowStrategy.Default): Observable[B]

    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.

    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.

    parallelism

    is the maximum number of tasks that can be executed in parallel, over which the source starts being back-pressured

    f

    is the mapping function that produces tasks to execute in parallel, which will eventually produce events for the resulting observable stream

    See also

    mapParallelUnordered for a variant that does not preserve order which may lead to faster execution times

    mapEval for serial execution

  158. final def mapParallelOrderedF[F[_], B](parallelism: Int)(f: (A) => F[B])(implicit os: OverflowStrategy[B] = OverflowStrategy.Default, F: TaskLike[F]): Observable[B]

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

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

    is the maximum number of tasks that can be executed in parallel, over which the source starts being back-pressured

    f

    is the mapping function that produces tasks to execute in parallel, which will eventually produce events for the resulting observable stream

    See also

    mapParallelUnorderedF for a variant that does not preserve order which may lead to faster execution times

    mapEvalF for serial execution

  159. final def mapParallelUnordered[B](parallelism: Int)(f: (A) => Task[B])(implicit os: OverflowStrategy[B] = OverflowStrategy.Default): Observable[B]

    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.

    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.

    parallelism

    is the maximum number of tasks that can be executed in parallel, over which the source starts being back-pressured

    f

    is the mapping function that produces tasks to execute in parallel, which will eventually produce events for the resulting observable stream

    See also

    mapParallelOrdered for a variant that does preserve order

    mapEval for serial execution

  160. final def mapParallelUnorderedF[F[_], B](parallelism: Int)(f: (A) => F[B])(implicit os: OverflowStrategy[B] = OverflowStrategy.Default, F: TaskLike[F]): Observable[B]

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

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

    is the maximum number of tasks that can be executed in parallel, over which the source starts being back-pressured

    f

    is the mapping function that produces tasks to execute in parallel, which will eventually produce events for the resulting observable stream

    See also

    mapParallelOrdered for a variant that does preserve order

    mapEval for serial execution

  161. final def materialize: Observable[Notification[A]]

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

  162. final def max[AA >: A](implicit A: Order[AA]): Observable[AA]

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

    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
    A

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the maximum element of the source stream, relative to the defined Order

    See also

    maxL for the version that returns a Task instead of an observable.

  163. final def maxBy[K](key: (A) => K)(implicit K: Order[K]): Observable[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.

    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
    key

    is the function that returns the key for which the given ordering is defined

    K

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the maximum element of the source stream, relative to its key generated by the given function and the given ordering

    See also

    maxByL for the version that returns a Task instead of an observable.

  164. final def maxByL[K](key: (A) => K)(implicit K: Order[K]): Task[Option[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.

    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
    key

    is the function that returns the key for which the given ordering is defined

    K

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the maximum element of the source stream, relative to its key generated by the given function and the given ordering

    See also

    maxBy for the version that returns an observable instead of a Task.

  165. final def maxL[AA >: A](implicit A: Order[AA]): Task[Option[AA]]

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

    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
    A

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the maximum element of the source stream, relative to the defined Order

    See also

    maxF for the version that returns an observable instead of a Task.

  166. final def merge[B](implicit ev: <:<[A, Observable[B]], os: OverflowStrategy[B] = OverflowStrategy.Default[B]): Observable[B]

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

    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
    

    returns

    an observable containing the merged events of all streams created by the source

    Note

    this operation needs to do buffering and by not specifying an OverflowStrategy, the default strategy is being used.

  167. final def mergeDelayErrors[B](implicit ev: <:<[A, Observable[B]], os: OverflowStrategy[B] = OverflowStrategy.Default): Observable[B]

    This version is reserving onError notifications until all of the observables complete and only then passing the issued errors(s) downstream.

    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.

    returns

    an observable containing the merged events of all streams created by the source

    Note

    this operation needs to do buffering and by not specifying an OverflowStrategy, the default strategy is being used.

  168. final def mergeMap[B](f: (A) => Observable[B])(implicit os: OverflowStrategy[B] = OverflowStrategy.Default): Observable[B]

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

    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
    

    f

    is a generator for the streams that will get merged

    returns

    an observable that emits the result of applying the transformation function to each item emitted by the source observable and merging the results of the observables obtained from this transformation.

  169. final def mergeMapDelayErrors[B](f: (A) => Observable[B])(implicit os: OverflowStrategy[B] = OverflowStrategy.Default): Observable[B]

    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.

    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.

    f

    is a generator for the streams that will get merged

    returns

    an observable that emits the result of applying the transformation function to each item emitted by the source observable and merging the results of the observables obtained from this transformation.

  170. final def min[AA >: A](implicit A: Order[AA]): Observable[AA]

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

    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
    A

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the minimum element of the source stream, relative to the defined Order

    See also

    minL for the version that returns a Task instead of an observable.

  171. final def minBy[K](key: (A) => K)(implicit K: Order[K]): Observable[A]

    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.

    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
    key

    is the function that returns the key for which the given ordering is defined

    K

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the minimum element of the source stream, relative to its key generated by the given function and the given ordering

  172. final def minByL[K](key: (A) => K)(implicit K: Order[K]): Task[Option[A]]

    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.

    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
    key

    is the function that returns the key for which the given ordering is defined

    K

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the minimum element of the source stream, relative to its key generated by the given function and the given ordering

  173. final def minL[AA >: A](implicit A: Order[AA]): Task[Option[AA]]

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

    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
    A

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the minimum element of the source stream, relative to the defined Order

    See also

    minF for the version that returns an observable instead of a Task.

  174. final def multicast[B >: A, R](pipe: Pipe[B, R])(implicit s: Scheduler): ConnectableObservable[R]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

    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.

    Annotations
    @UnsafeBecauseImpure()
  175. final def ne(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  176. final def nonEmpty: Observable[Boolean]

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

  177. final def nonEmptyL: Task[Boolean]

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

  178. final def notify(): Unit
    Definition Classes
    AnyRef
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  179. final def notifyAll(): Unit
    Definition Classes
    AnyRef
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  180. final def observeOn[B >: A](s: Scheduler, os: OverflowStrategy[B]): Observable[B]

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

    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.

    s

    is the alternative Scheduler reference to use for observing events

    os

    is the OverflowStrategy to apply to the underlying buffer

    See also

    observeOn(Scheduler) for the version that does not take an OverflowStrategy parameter.

  181. final def observeOn(s: Scheduler): Observable[A]

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

    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.

    s

    is the alternative Scheduler reference to use for observing events

    See also

    observeOn(Scheduler, OverflowStrategy) for the version that allows customizing the OverflowStrategy being used by the underlying buffer.

  182. final def onCancelTriggerError: Observable[A]

    If the connection is cancelled then trigger a CancellationException.

    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.

  183. final def onErrorFallbackTo[B >: A](that: Observable[B]): Observable[B]

    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.

    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.

    that

    is a backup sequence that's being subscribed in case the source terminates with an error.

  184. final def onErrorHandle[B >: A](f: (Throwable) => B): Observable[B]

    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.

    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.

    f

    - a function that matches errors with a backup element that is emitted when the source throws an error.

  185. final def onErrorHandleWith[B >: A](f: (Throwable) => Observable[B]): Observable[B]

    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.

    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.

    f

    is a function that matches errors with a backup throwable that is subscribed when the source throws an error.

  186. final def onErrorRecover[B >: A](pf: PartialFunction[Throwable, B]): Observable[B]

    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.

    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.

    pf

    is a function that matches errors with a backup element that is emitted when the source throws an error.

  187. final def onErrorRecoverWith[B >: A](pf: PartialFunction[Throwable, Observable[B]]): Observable[B]

    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.

    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.

    pf

    is a function that matches errors with a backup throwable that is subscribed when the source throws an error.

  188. final def onErrorRestart(maxRetries: Long): Observable[A]

    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.

    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.

  189. final def onErrorRestartIf(p: (Throwable) => Boolean): Observable[A]

    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.

    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.

  190. final def onErrorRestartUnlimited: Observable[A]

    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.

    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.

  191. final def pipeThrough[I >: A, B](pipe: Pipe[I, B]): Observable[B]

    Given a Pipe, transform the source observable with it.

  192. final def pipeThroughSelector[S >: A, B, R](pipe: Pipe[S, B], f: (Observable[B]) => Observable[R]): Observable[R]

    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.

    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._
    implicit val os: OverflowStrategy[Nothing] = OverflowStrategy.Default
    
    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.

    pipe

    is the Pipe used to transform the source into a multicast (hot) observable that can be shared in the selector function

    f

    is a selector function that can use the multicasted source sequence as many times as needed, without causing multiple subscriptions to the source sequence. Observers to the given source will receive all notifications of the source from the time of the subscription forward.

  193. final def prepend[B >: A](elem: B): Observable[B]

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

  194. final def publish(implicit s: Scheduler): ConnectableObservable[A]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

    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.

    Annotations
    @UnsafeBecauseImpure()
  195. final def publishLast(implicit s: Scheduler): ConnectableObservable[A]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

    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.

    Annotations
    @UnsafeBecauseImpure()
  196. final def publishSelector[R](f: (Observable[A]) => Observable[R]): Observable[R]

    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.

    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._
    implicit val os: OverflowStrategy[Nothing] = OverflowStrategy.Default
    
    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.

    f

    is a selector function that can use the multicasted source sequence as many times as needed, without causing multiple subscriptions to the source sequence. Observers to the given source will receive all notifications of the source from the time of the subscription forward.

    See also

    pipeThroughSelector for a version that allows specifying a type of underlying Subject.

  197. final def reduce[B >: A](op: (B, B) => B): Observable[B]

    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.

  198. final def repeat: Observable[A]

    Repeats the items emitted by the source continuously.

    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.

  199. final def replay(bufferSize: Int)(implicit s: Scheduler): ConnectableObservable[A]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

    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.

    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.

    bufferSize

    is the size of the buffer limiting the number of items that can be replayed (on overflow the head starts being dropped)

    Annotations
    @UnsafeBecauseImpure()
  200. final def replay(implicit s: Scheduler): ConnectableObservable[A]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

    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.

    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.

    Annotations
    @UnsafeBecauseImpure()
  201. final def restartUntil(p: (A) => Boolean): Observable[A]

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

  202. final def runAsyncGetFirst(implicit s: Scheduler, opts: Options = defaultOptions): CancelableFuture[Option[A]]

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

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

    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.

    Annotations
    @UnsafeBecauseImpure()
  203. final def runAsyncGetLast(implicit s: Scheduler, opts: Options = defaultOptions): CancelableFuture[Option[A]]

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

    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.

    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.

    Annotations
    @UnsafeBecauseImpure()
  204. final def sample(period: FiniteDuration): Observable[A]

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

    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.

    Usage:

    import scala.concurrent.duration._
    
    // emits 3, 8, 10 in 1 second intervals
    Observable.fromIterable(0 to 10)
      // without delay, it would return only 10
      .delayOnNext(200.millis)
      .sample(1.second)
    period

    the timespan at which sampling occurs

    See also

    sampleBy for fine control

    sampleRepeated for repeating the last value on silence

    throttle for a version that allows to specify number of elements processed by a period and does not drop any elements

  205. final def sampleBy[B](sampler: Observable[B]): Observable[A]

    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.

    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.

    sampler

    - the observable to use for sampling the source

    See also

    sample for periodic sampling

    sampleRepeatedBy for repeating the last value on silence

  206. final def sampleRepeated(period: FiniteDuration): Observable[A]

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

    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.

    period

    the timespan at which sampling occurs

    See also

    sample for a variant that doesn't repeat the last value on silence

    sampleRepeatedBy for fine control

  207. final def sampleRepeatedBy[B](sampler: Observable[B]): Observable[A]

    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.

    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.

    sampler

    - the Observable to use for sampling the source Observable

    See also

    sampleBy for a variant that doesn't repeat the last value on silence

    sampleRepeated for a periodic sampling

  208. final def scan[S](seed: => S)(op: (S, A) => S): Observable[S]

    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.

    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 foldLeft, but emits the state on each step. Useful for modeling finite state machines.

    See also

    scan0 for the version that emits seed element at the beginning

  209. final def scan0[S](seed: => S)(op: (S, A) => S): Observable[S]

    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.

    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.

    This is a version of scan that emits seed element at the beginning, similar to scanLeft on Scala collections

  210. final def scanEval[S](seed: Task[S])(op: (S, A) => Task[S]): Observable[S]

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

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

    Similar with scan, but this can suspend and evaluate side effects with Task, thus allowing for asynchronous data processing.

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

    Example showing how state can be evolved and acted upon:

    import monix.eval.Task
    
    sealed trait State[+A] { def count: Int }
    case object Init extends State[Nothing] { def count = 0 }
    case class Current[A](current: Option[A], count: Int)
      extends State[A]
    
    case class Person(id: Int, name: String)
    
    // TODO: to implement!
    def requestPersonDetails(id: Int): Task[Option[Person]] =
      Task.raiseError(new NotImplementedError)
    
    // TODO: to implement
    val source: Observable[Int] =
      Observable.raiseError(new NotImplementedError)
    
    // Initial state
    val seed = Task.pure(Init : State[Person])
    
    val scanned = source.scanEval(seed) { (state, id) =>
      requestPersonDetails(id).map { person =>
        state match {
          case Init =>
            Current(person, 1)
          case Current(_, count) =>
            Current(person, count + 1)
        }
      }
    }
    
    val filtered = scanned
      .takeWhile(_.count < 10)
      .collect { case Current(a, _) => a }
    seed

    is the initial state

    op

    is the function that evolves the current state

    returns

    a new observable that emits all intermediate states being resulted from applying the given function

    See also

    scanEval0 for the version that emits seed element at the beginning

    scan for the version that does not require using Task in the provided operator

  211. final def scanEval0[S](seed: Task[S])(op: (S, A) => Task[S]): Observable[S]

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

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

    This is a version of scanEval that emits seed element at the beginning.

  212. final def scanEval0F[F[_], S](seed: F[S])(op: (S, A) => F[S])(implicit F: TaskLike[F], A: Applicative[F]): Observable[S]

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

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

    This is a version of scanEvalF that emits seed element at the beginning, similar to scanLeft on Scala collections

  213. final def scanEvalF[F[_], S](seed: F[S])(op: (S, A) => F[S])(implicit F: TaskLike[F]): Observable[S]

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

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

    Similar with scan, but this can suspend and evaluate side effects with an F[_] data type that implements the cats.effect.Effect type class, thus allowing for lazy or asynchronous data processing.

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

    Example showing how state can be evolved and acted upon:

    // Using cats.effect.IO for evaluating our side effects
    import cats.effect.IO
    
    sealed trait State[+A] { def count: Int }
    case object Init extends State[Nothing] { def count = 0 }
    case class Current[A](current: Option[A], count: Int)
      extends State[A]
    
    case class Person(id: Int, name: String)
    
    // TODO: to implement!
    def requestPersonDetails(id: Int): IO[Option[Person]] =
      IO.raiseError(new NotImplementedError)
    
    // TODO: to implement
    val source: Observable[Int] =
      Observable.raiseError(new NotImplementedError)
    
    // Initial state
    val seed = IO.pure(Init : State[Person])
    
    val scanned = source.scanEvalF(seed) { (state, id) =>
      requestPersonDetails(id).map { person =>
        state match {
          case Init =>
            Current(person, 1)
          case Current(_, count) =>
            Current(person, count + 1)
        }
      }
    }
    
    val filtered = scanned
      .takeWhile(_.count < 10)
      .collect { case Current(a, _) => a }
    seed

    is the initial state

    op

    is the function that evolves the current state

    F

    is the cats.effect.Effect type class implementation for type F, which controls the evaluation. F can be a data type such as monix.eval.Task or cats.effect.IO, which implement Effect.

    returns

    a new observable that emits all intermediate states being resulted from applying the given function

    See also

    scanEval0F for the version that emits seed element at the beginning

    scan for the synchronous, non-lazy version, or scanEval for the Task-specialized version.

  214. final def scanMap[B](f: (A) => B)(implicit B: Monoid[B]): Observable[B]

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    Equivalent with scan applied with the given cats.Monoid, so given our f mapping function returns a B, this law holds:

    val B = implicitly[Monoid[B]]
    
    stream.scanMap(f) <-> stream.scan(B.empty)(B.combine)
    

    Example:

    import cats.implicits._
    
    // Yields 2, 6, 12, 20, 30, 42
    val stream = Observable(1, 2, 3, 4, 5, 6).scanMap(x => x * 2)
    f

    is the mapping function applied to every incoming element of this Observable before folding using Monoid[B].combine

    returns

    a new Observable that emits all intermediate states being resulted from applying Monoid[B].combine function

  215. final def scanMap0[B](f: (A) => B)(implicit B: Monoid[B]): Observable[B]

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    This is a version of scanMap that emits seed element at the beginning.

  216. final def share(implicit s: Scheduler): Observable[A]

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

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

    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.

    Annotations
    @UnsafeBecauseImpure()
  217. final def startWith[B >: A](elems: Seq[B]): Observable[B]

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

  218. final def subscribe(nextFn: (A) => Future[Ack], errorFn: (Throwable) => Unit, completedFn: () => Unit)(implicit s: Scheduler): Cancelable

    Subscribes to the stream.

    Subscribes to the stream.

    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

    a subscription that can be used to cancel the streaming.

    Annotations
    @UnsafeBecauseImpure()
    See also

    consumeWith for another way of consuming observables

  219. final def subscribe(nextFn: (A) => Future[Ack])(implicit s: Scheduler): Cancelable

    Subscribes to the stream.

    Subscribes to the stream.

    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

    a subscription that can be used to cancel the streaming.

    Annotations
    @UnsafeBecauseImpure()
    See also

    consumeWith for another way of consuming observables

  220. final def subscribe()(implicit s: Scheduler): Cancelable

    Subscribes to the stream.

    Subscribes to the stream.

    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

    a subscription that can be used to cancel the streaming.

    Annotations
    @UnsafeBecauseImpure()
    See also

    consumeWith for another way of consuming observables

  221. final def subscribe(nextFn: (A) => Future[Ack], errorFn: (Throwable) => Unit)(implicit s: Scheduler): Cancelable

    Subscribes to the stream.

    Subscribes to the stream.

    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

    a subscription that can be used to cancel the streaming.

    Annotations
    @UnsafeBecauseImpure()
    See also

    consumeWith for another way of consuming observables

  222. final def subscribe(subscriber: Subscriber[A]): Cancelable

    Subscribes to the stream.

    Subscribes to the stream.

    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

    a subscription that can be used to cancel the streaming.

    Annotations
    @UnsafeBecauseImpure()
    See also

    consumeWith for another way of consuming observables

  223. final def subscribe(observer: Observer[A])(implicit s: Scheduler): Cancelable

    Subscribes to the stream.

    Subscribes to the stream.

    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

    a subscription that can be used to cancel the streaming.

    Annotations
    @UnsafeBecauseImpure()
    See also

    consumeWith for another way of consuming observables

  224. final def subscribeOn(scheduler: Scheduler): Observable[A]

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

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

    This is different from executeOn because the given scheduler is only used to start the subscription, but does not override the default Scheduler.

  225. final def sum[AA >: A](implicit A: Numeric[AA]): Observable[AA]

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

  226. final def sumL[B >: A](implicit B: Numeric[B]): Task[B]

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

  227. final def switch[B](implicit ev: <:<[A, Observable[B]]): Observable[B]

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

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

    Similar with flatten, however the source isn't back-pressured when emitting new events. Instead new events being emitted are cancelling the active child observables.

    Equivalence with switchMap

    The switch operation can be expressed in terms of switchMap, as we have this equivalence:

    stream.switch <-> stream.switchMap(x => x)

    See also

    the description of switchMap for an example.

  228. final def switchIfEmpty[B >: A](backup: Observable[B]): Observable[B]

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

  229. final def switchMap[B](f: (A) => Observable[B]): Observable[B]

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

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

    Similar with concatMap, however the source isn't back-pressured when emitting new events. Instead new events being emitted are cancelling the active child observables.

    Example

    The switchMap can express a lot of cool, time-based operations. For example we can express debounce in terms of switchMap:

    import scala.concurrent.duration._
    
    def debounce[A](stream: Observable[A], d: FiniteDuration): Observable[A] =
      stream.switchMap { x =>
        Observable.now(x).delayExecution(d)
      }
    f

    is a generator for the streams that are being merged

  230. final def synchronized[T0](arg0: => T0): T0
    Definition Classes
    AnyRef
  231. final def tail: Observable[A]

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

  232. final def take(n: Long): Observable[A]

    Selects the first n elements (from the start).

    Selects the first n elements (from the start).

    n

    the number of elements to take

    returns

    a new Observable that emits only the first n elements from the source

  233. final def takeByTimespan(timespan: FiniteDuration): Observable[A]

    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 the events of the source, only for the specified timestamp, after which it completes.

    timespan

    the window of time during which the new Observable is allowed to emit the events of the source

  234. final def takeEveryNth(n: Int): Observable[A]

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

  235. final def takeLast(n: Int): Observable[A]

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

    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.

  236. final def takeUntil(trigger: Observable[Any]): Observable[A]

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

    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.

    trigger

    is an observable that will cancel the streaming as soon as it emits an event

  237. final def takeUntilEval(trigger: Task[_]): Observable[A]

    Version of takeUntil that can work with a trigger expressed by a monix.eval.Task

    Version of takeUntil that can work with a trigger expressed by a monix.eval.Task

    trigger

    task that will cancel the stream as soon as it completes.

    See also

    takeUntil for version that works with Observable.

    takeUntilEvalF for version that works with generic F[_] powered by monix.eval.TaskLike.

  238. final def takeUntilEvalF[F[_], B](trigger: F[B])(implicit taskLike: TaskLike[F]): Observable[A]

    Version of takeUntil that can work with a trigger expressed by a generic F[_] provided an implicit monix.eval.TaskLike exists.

    Version of takeUntil that can work with a trigger expressed by a generic F[_] provided an implicit monix.eval.TaskLike exists.

    trigger

    operation that will cancel the stream as soon as it completes.

    See also

    takeUntil for version that works with Observable.

    takeUntilEval for version that works with monix.eval.Task.

  239. final def takeWhile(p: (A) => Boolean): Observable[A]

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

  240. final def takeWhileInclusive(p: (A) => Boolean): Observable[A]

    Takes longest prefix of elements that satisfy the given predicate, inclusive of the value that caused predicate to return false and returns a new Observable that emits those elements.

  241. final def takeWhileNotCanceled(c: BooleanCancelable): Observable[A]

    Takes longest prefix of elements while given BooleanCancelable is not canceled and returns a new Observable that emits those elements.

  242. final def throttle(period: FiniteDuration, n: Int): Observable[A]

    Returns an Observable that emits maximum n items per given period.

    Returns an Observable that emits maximum n items per given period.

    Unlike Observable!.throttleLast and Observable!.throttleFirst it does not discard any elements.

    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:

    import scala.concurrent.duration._
    
    // emits two items per second
    Observable.fromIterable(0 to 10)
      .throttle(1.second, 2)
    period

    time that has to pass before emiting new items

    n

    maximum number of items emitted per given period

  243. final def throttleFirst(interval: FiniteDuration): Observable[A]

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

    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.

    Usage:

    import scala.concurrent.duration._
    
    // emits 0, 5, 10 in 1 second intervals
    Observable.fromIterable(0 to 10)
      // without delay, it would return only 0
      .delayOnNext(200.millis)
      .throttleFirst(1.second)
    interval

    time to wait before emitting another item after emitting the last item

    See also

    throttle for a version that allows to specify number of elements processed by a period and does not drop any elements

  244. final def throttleLast(period: FiniteDuration): Observable[A]

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

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

    Alias for sample.

    Usage:

    import scala.concurrent.duration._
    
    // emits 3, 8, 10 in 1 second intervals
    Observable.fromIterable(0 to 10)
      // without delay, it would return only 10
      .delayOnNext(200.millis)
      .throttleLast(1.second)
    period

    duration of windows within which the last item emitted by the source Observable will be emitted

    See also

    throttle for a version that allows to specify number of elements processed by a period and does not drop any elements

  245. final def throttleLatest(period: FiniteDuration, emitLast: Boolean): Observable[A]

    Emit first element emitted by the source and then emit the most recent items emitted by the source within periodic time intervals.

    Emit first element emitted by the source and then emit the most recent items emitted by the source within periodic time intervals. Usage:

    import scala.concurrent.duration._
    
    // emits 0 after 200 ms and then 4,9 in 1 sec intervals and 10 after the observable completes
    Observable.fromIterable(0 to 10)
      // without delay, it would return only 0, 10
      .delayOnNext(200.millis)
      .throttleLatest(1.second, true)
    period

    duration of windows within which the last item emitted by the source Observable will be emitted

    emitLast

    if true last element will be emitted when source completes no matter if interval has passed or not

  246. final def throttleWithTimeout(timeout: FiniteDuration): Observable[A]

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

    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.

    timeout

    the length of the window of time that must pass after the emission of an item from the source observable in which that observable emits no items in order for the item to be emitted by the resulting observable

    See also

    echoOnce for a similar operator that also mirrors the source observable

  247. final def timeoutOnSlowDownstream(timeout: FiniteDuration): Observable[A]

    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.

    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.

    timeout

    maximum duration for onNext.

  248. final def timeoutOnSlowDownstreamTo[B >: A](timeout: FiniteDuration, backup: Observable[B]): Observable[B]

    Returns an observable that mirrors the source but applies a timeout for each onNext message.

    Returns an observable that mirrors the source but applies a timeout for each onNext message. If downstream subscriber takes more time than the given timespan to process an onNext message, the source is terminated and downstream gets subscribed to the given backup.

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

    timeout

    maximum duration for onNext.

    backup

    alternative data source to subscribe to on timeout.

  249. final def timeoutOnSlowUpstream(timeout: FiniteDuration): Observable[A]

    Returns an observable that mirrors the source but applies a timeout for each emitted item by the upstream.

    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.

    timeout

    maximum duration between emitted items before a timeout occurs (ignoring the time it takes to process onNext)

  250. final def timeoutOnSlowUpstreamTo[B >: A](timeout: FiniteDuration, backup: Observable[B]): Observable[B]

    Returns an observable that mirrors the source but applies a timeout for each emitted item by the upstream.

    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.

    timeout

    maximum duration between emitted items before a timeout occurs (ignoring the time it takes to process onNext)

    backup

    is the alternative data source to subscribe to on timeout

  251. final def toListL: Task[List[A]]

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

    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.

  252. final def toReactivePublisher[B >: A](implicit s: Scheduler): Publisher[B]

    Converts this Observable into an org.reactivestreams.Publisher.

    Converts this Observable into an org.reactivestreams.Publisher.

    Meant for interoperability with other Reactive Streams implementations.

    Usage sample:

    import monix.eval.Task
    import monix.execution.rstreams.SingleAssignSubscription
    import org.reactivestreams.{Publisher, Subscriber, Subscription}
    
    def sum(source: Publisher[Int], requestSize: Int): Task[Long] =
      Task.create { (_, cb) =>
        val sub = SingleAssignSubscription()
    
        source.subscribe(new Subscriber[Int] {
          private[this] var requested = 0L
          private[this] var sum = 0L
    
          def onSubscribe(s: Subscription): Unit = {
            sub := s
            requested = requestSize
            s.request(requestSize)
          }
    
          def onNext(t: Int): Unit = {
            sum += t
            if (requestSize != Long.MaxValue) requested -= 1
    
            if (requested <= 0) {
              requested = requestSize
              sub.request(requestSize)
            }
          }
    
          def onError(t: Throwable): Unit =
            cb.onError(t)
          def onComplete(): Unit =
            cb.onSuccess(sum)
        })
    
        // Cancelable that can be used by Task
        sub
      }
    
    import monix.execution.Scheduler.Implicits.global
    val pub = Observable(1, 2, 3, 4).toReactivePublisher
    
    // Yields 10
    sum(pub, requestSize = 128)

    See the Reactive Streams protocol for details.

  253. def toString(): String
    Definition Classes
    AnyRef → Any
  254. def transform[B](transformer: Transformer[A, B]): Observable[B]

    Transforms the source using the given transformer function.

  255. final def uncancelable: Observable[A]

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

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

    Example

    import scala.concurrent.duration._
    
    Observable.eval(println("Hello!"))
      .delayExecution(10.seconds)
      .uncancelable

    The created observable, after subscribe, will print "Hello!" even if cancellation is attempted.

  256. final def unsafeMulticast[B >: A, R](processor: Subject[B, R])(implicit s: Scheduler): ConnectableObservable[R]

    Converts this observable into a multicast observable, useful for turning a cold observable into a hot one (i.e.

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

    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.

    Annotations
    @UnsafeProtocol() @UnsafeBecauseImpure()
  257. final def unsafeSubscribeFn(observer: Observer[A])(implicit s: Scheduler): Cancelable

    Given an observer and a scheduler for managing async boundaries, subscribes to this observable for events.

    Given an observer and a scheduler for managing async boundaries, subscribes to this observable for events.

    Helper for calling the abstract method.

    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.

    Annotations
    @UnsafeProtocol() @UnsafeBecauseImpure()
  258. final def wait(arg0: Long, arg1: Int): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException])
  259. final def wait(arg0: Long): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException]) @native()
  260. final def wait(): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException])
  261. def whileBusyAggregateEvents[S](seed: (A) => S)(aggregate: (S, A) => S): Observable[S]

    Conflates events when a downstream is slower than the upstream.

    Conflates events when a downstream is slower than the upstream.

    Emits: Immediately when an element is received if the downstream is waiting for elements. Otherwise emits when the downstream stops backpressuring and there is a conflated element available. Back pressures: Never (conflates instead)

    Usage:

    import scala.concurrent.duration._
    import cats.data.Chain
    
    // Emits [0], [1, 2], [3, 4]
    Observable.range(0, 5)
      .throttle(1.second, 1)
      .whileBusyAggregateEvents(Chain.apply(_)){ case (chain, ele) => chain.append(ele) }
      .throttle(2.seconds, 1)
  262. final def whileBusyBuffer[B >: A](overflowStrategy: Synchronous[B]): Observable[B]

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

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

    overflowStrategy

    - the overflow strategy used for buffering, which specifies what to do in case we're dealing with a slow consumer - should an unbounded buffer be used, should back-pressure be applied, should the pipeline drop newer or older events, should it drop the whole buffer? See OverflowStrategy for more details.

  263. final def whileBusyDropEvents: Observable[A]

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

  264. final def whileBusyDropEventsAndSignal[B >: A](onOverflow: (Long) => B): Observable[B]

    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.

    onOverflow

    - a function that is used for signaling a special event used to inform the consumers that an overflow event happened, function that receives the number of dropped events as a parameter (see OverflowStrategy.Evicted)

  265. final def whileBusyReduceEvents[B >: A](op: (B, B) => B): Observable[B]

    Reduces elements when a downstream is slower than the upstream.

    Reduces elements when a downstream is slower than the upstream.

    Emits: Immediately when an element is received if the downstream is waiting for elements. Otherwise emits when the downstream stops backpressuring and there is a reduced element available. Back pressures: Never (reduces instead)

    Usage:

    import scala.concurrent.duration._
    import cats.data.Chain
    
    // Emits 0, 3 (1+2), 7 (3+4)
    Observable.range(0, 5)
      .throttle(1.second, 1)
      .whileBusyReduceEvents(_ + _)
      .throttle(2.seconds, 1)
  266. final def withFilter(p: (A) => Boolean): Observable[A]

    Alias to filter to support syntax in for comprehension, i.e.

    Alias to filter to support syntax in for comprehension, i.e.

    Example:

    case class Person(age: Long)
    
    val peopleObservable: Observable[Person] =
      Observable.range(1, 100).map(Person.apply)
    
    for {
      adult <- peopleObservable if adult.age >= 18
    } yield adult
  267. final def withLatestFrom[B, R](other: Observable[B])(f: (A, B) => R): Observable[R]

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

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

    Visual Example

    stream1: 1 - - 2 - - 3 - 4 - -
    stream2: 1 - - 2 - 3 - - - - 4
    
    result: (1, 1), (2, 2), (3, 3), (4, 3)
    

    other

    is an observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  268. final def withLatestFrom2[B1, B2, R](o1: Observable[B1], o2: Observable[B2])(f: (A, B1, B2) => R): Observable[R]

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

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

    o1

    is the first observable that gets paired with the source

    o2

    is the second observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  269. final def withLatestFrom3[B1, B2, B3, R](o1: Observable[B1], o2: Observable[B2], o3: Observable[B3])(f: (A, B1, B2, B3) => R): Observable[R]

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

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

    o1

    is the first observable that gets paired with the source

    o2

    is the second observable that gets paired with the source

    o3

    is the third observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  270. final def withLatestFrom4[B1, B2, B3, B4, R](o1: Observable[B1], o2: Observable[B2], o3: Observable[B3], o4: Observable[B4])(f: (A, B1, B2, B3, B4) => R): Observable[R]

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

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

    o1

    is the first observable that gets paired with the source

    o2

    is the second observable that gets paired with the source

    o3

    is the third observable that gets paired with the source

    o4

    is the fourth observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  271. final def withLatestFrom5[B1, B2, B3, B4, B5, R](o1: Observable[B1], o2: Observable[B2], o3: Observable[B3], o4: Observable[B4], o5: Observable[B5])(f: (A, B1, B2, B3, B4, B5) => R): Observable[R]

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

    Combines the elements emitted by the source with the latest elements emitted by five 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).

    o1

    is the first observable that gets paired with the source

    o2

    is the second observable that gets paired with the source

    o3

    is the third observable that gets paired with the source

    o4

    is the fourth observable that gets paired with the source

    o5

    is the fifth observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  272. final def withLatestFrom6[B1, B2, B3, B4, B5, B6, R](o1: Observable[B1], o2: Observable[B2], o3: Observable[B3], o4: Observable[B4], o5: Observable[B5], o6: Observable[B6])(f: (A, B1, B2, B3, B4, B5, B6) => R): Observable[R]

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

    Combines the elements emitted by the source with the latest elements emitted by six 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).

    o1

    is the first observable that gets paired with the source

    o2

    is the second observable that gets paired with the source

    o3

    is the third observable that gets paired with the source

    o4

    is the fourth observable that gets paired with the source

    o5

    is the fifth observable that gets paired with the source

    o6

    is the sixth observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  273. final def zip[B](other: Observable[B]): Observable[(A, B)]

    Creates a new observable from this observable and another given observable by combining their items in pairs in a strict sequence.

    Creates a new observable from this observable and another given observable by combining their items in pairs in a strict sequence.

    So the first item emitted by the new observable will be the tuple of the first items emitted by each of the source observables; the second item emitted by the new observable will be a tuple with the second items emitted by each of those observables; and so forth.

    Visual Example

    stream1: 1 - - 2 - - 3 - 4 - -
    stream2: 1 - - 2 - 3 - - - - 4
    
    result: (1, 1), (2, 2), (3, 3), (4, 4)
    

    See combineLatest for a more relaxed alternative that doesn't combine items in strict sequence.

    other

    is an observable that gets paired with the source

    returns

    a new observable sequence that emits the paired items of the source observables

  274. final def zipMap[B, R](other: Observable[B])(f: (A, B) => R): Observable[R]

    Creates a new observable from this observable and another given observable by combining their items in pairs in a strict sequence.

    Creates a new observable from this observable and another given observable by combining their items in pairs in a strict sequence.

    So the first item emitted by the new observable will be the result of the function applied to the first item emitted by each of the source observables; the second item emitted by the new observable will be the result of the function applied to the second item emitted by each of those observables; and so forth.

    Visual Example

    stream1: 1 - - 2 - - 3 - 4 - -
    stream2: 1 - - 2 - 3 - - - - 4
    
    result: (1, 1), (2, 2), (3, 3), (4, 4)
    

    See combineLatestMap for a more relaxed alternative that doesn't combine items in strict sequence.

    other

    is an observable that gets paired with the source

    f

    is a mapping function over the generated pairs

  275. final def zipWithIndex: Observable[(A, Long)]

    Zips the emitted elements of the source with their indices.

Deprecated Value Members

  1. def finalize(): Unit
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.Throwable]) @Deprecated
    Deprecated

Inherited from Serializable

Inherited from AnyRef

Inherited from Any

Ungrouped