Creates a combined observable from 2 source observables.

This operator behaves in a similar way to zip2, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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)

Creates a combined observable from 3 source observables.

This operator behaves in a similar way to zip3, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 4 source observables.

This operator behaves in a similar way to zip4, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 5 source observables.

This operator behaves in a similar way to zip5, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 6 source observables.

This operator behaves in a similar way to zip6, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 2 source observables.

This operator behaves in a similar way to zipMap2, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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)

Creates a combined observable from 3 source observables.

This operator behaves in a similar way to zipMap3, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 4 source observables.

This operator behaves in a similar way to zipMap4, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 5 source observables.

This operator behaves in a similar way to zipMap5, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates a combined observable from 6 source observables.

This operator behaves in a similar way to zipMap6, but while zip emits items only when all of the zipped source observables have emitted a previously unzipped item, combine 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).

Creates an observable from a function that receives a concurrent and safe Subscriber.Sync.

This builder represents the safe way of building observables from data-sources that cannot be back-pressured.

Given a list of source Observables, emits all of the items from the first of these Observables to emit an item or to complete, and cancel the rest.

Visual Example

stream1: - - 1 1 1 - 1 - 1 - -
stream2: - - - - - 2 2 2 2 2 2

result: - - 1 1 1 - 1 - 1 - -

Converts to Observable from any F[_] that has an ObservableLike instance.

Supported types includes, but is not necessarily limited to:

• cats.Eval
• cats.effect.IO
• cats.effect.SyncIO
• cats.effect.Effect (Async)
• cats.effect.ConcurrentEffect
• org.reactivestreams.Publisher
• monix.eval.Coeval
• monix.eval.Task
• scala.Either
• scala.util.Try
• scala.concurrent.Future

Version of fromAsyncStateAction 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
• ...

Safely converts a java.io.Reader into an observable that will emit Array[Char] elements.

Compared with fromCharsReaderUnsafe, this version:

• is referentially transparent, the input being a "generator" powered by monix.eval.Task
• automatically forks execution on subscription to ensure that the current thread isn't blocked by the ensuing blocking I/O
• ensures that the input stream is closed on completion, failure or cancellation

Version of fromCharsReader 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
• ...

Converts a java.io.Reader into an observable that will emit Array[Char] elements.

UNSAFE WARNING: this is an unsafe function, because reading from a reader is a destructive process, also violating referential transparency. Therefore only a single subscriber is supported, the result being a single-subscriber observable. If multiple subscribers are attempted, all subscribers, except for the first one, will be terminated with a APIContractViolationException.

UNSAFE PROTOCOL: the created Observable does not close the given Reader. Usually it's the producer of a resource that needs to deallocate the resource.

This operation will start processing on the current thread (on subscribe()), so in order to not block, it might be better to also do an executeAsync, or you may want to use the AlwaysAsyncExecution model, which can be configured per Scheduler, see Scheduler.withExecutionModel, or per Observable, see Observable.executeWithModel.

Converts a Scala Future provided into an Observable.

If the created instance is a CancelableFuture, then it will be used for the returned Cancelable on subscribe.

Safely converts a java.io.InputStream into an observable that will emit Array[Byte] elements.

Compared with fromInputStreamUnsafe, this version:

• is referentially transparent, the input being a "generator" powered by monix.eval.Task
• automatically forks execution on subscription to ensure that the current thread isn't blocked by the ensuing blocking I/O
• ensures that the input stream is closed on completion, failure or cancellation

Version of fromInputStream 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
• ...

Converts a java.io.InputStream into an observable that will emit Array[Byte] elements.

UNSAFE WARNING: this is an unsafe function, because reading from an input stream is a destructive process, also violating referential transparency. Therefore only a single subscriber is supported, the result being a single-subscriber observable. If multiple subscribers are attempted, all subscribers, except for the first one, will be terminated with a APIContractViolationException.

UNSAFE PROTOCOL: the created Observable does not close the given InputStream. Usually it's the producer of a resource that needs to deallocate the resource.

This operation will start processing on the current thread (on subscribe()), so in order to not block, it might be better to also do an executeAsync, or you may want to use the AlwaysAsyncExecution model, which can be configured per Scheduler, see Scheduler.withExecutionModel, or per Observable, see Observable.executeWithModel.

Wraps a scala.Iterator into an Observable in the context of a cats.effect.Resource, which allows for specifying a finalizer.

Wraps a scala.Iterator into an Observable.

This function uses Task in order to suspend the creation of the Iterator, because reading from an Iterator is a destructive process. The Task is being used as a "factory", in pace of scala.Iterable.

Example:

import monix.eval.Task

Observable.fromIterator(Task(Iterator.from(1)))

Version of fromIterator 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
• ...

Converts any Iterator into an observable.

UNSAFE WARNING: reading from an Iterator is a destructive process. Therefore only a single subscriber is supported, the result being a single-subscriber observable. If multiple subscribers are attempted, all subscribers, except for the first one, will be terminated with a APIContractViolationException.

Safely converts a java.io.BufferedReader into an observable that will emit String elements corresponding to text lines from the input.

According to the specification of BufferedReader, a line is considered to be terminated by any one of a line feed (\n), a carriage return (\r), or a carriage return followed immediately by a linefeed.

Compared with fromLinesReaderUnsafe, this version:

• is referentially transparent, the input being a "generator" powered by monix.eval.Task
• automatically forks execution on subscription to ensure that the current thread isn't blocked by the ensuing blocking I/O
• ensures that the input stream is closed on completion, failure or cancellation

Version of fromLinesReader 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
• ...

Converts a java.io.BufferedReader into an observable that will emit String text lines from the input.

According to the specification of BufferedReader, a line is considered to be terminated by any one of a line feed (\n), a carriage return (\r), or a carriage return followed immediately by a linefeed.

UNSAFE WARNING: this is an unsafe function, because reading from a reader is a destructive process, also violating referential transparency. Therefore only a single subscriber is supported, the result being a single-subscriber observable. If multiple subscribers are attempted, all subscribers, except for the first one, will be terminated with a APIContractViolationException.

UNSAFE PROTOCOL: the created Observable does not close the given Reader. Usually it's the producer of a resource that needs to deallocate the resource.

Given a org.reactivestreams.Publisher, converts it into a Monix / Rx Observable.

See the Reactive Streams protocol that Monix implements.

Given a org.reactivestreams.Publisher, converts it into a Monix / Rx Observable.

See the Reactive Streams protocol that Monix implements.

Transforms any cats.effect.Resource into an Observable.

See the documentation for Resource.

import cats.effect.Resource
import monix.eval.Task
import java.io._

def openFileAsResource(file: File): Resource[Task, FileInputStream] =
  Resource.make(Task(new FileInputStream(file)))(h => Task(h.close()))

def openFileAsStream(file: File): Observable[FileInputStream] =
  Observable.fromResource(openFileAsResource(file))

This example would be equivalent with usage of Observable.resource:

def openFileAsResource2(file: File): Observable[FileInputStream] = {
  Observable.resource(Task(new FileInputStream(file)))(h => Task(h.close()))
}

This means that flatMap is safe to use:

def readBytes(file: File): Observable[Array[Byte]] =
  openFileAsStream(file).flatMap { in =>
    Observable.fromInputStreamUnsafe(in)
  }

Converts any Task into an Observable.

import monix.eval.Task

val task = Task.eval("Hello!")

Observable.fromTask(task)

Converts generic F[_] effects to Observable.

Currently supported data types:

• monix.eval.Task
• monix.eval.Coeval
• scala.concurrent.Future
• cats.effect.IO
• any cats.effect.Effect
• any cats.effect.ConcurrentEffect

Sample:

import cats.implicits._
import cats.effect.IO
import scala.concurrent.duration._
import monix.execution.Scheduler.global
import monix.catnap.SchedulerEffect

// Needed for IO.sleep
implicit val timer = SchedulerEffect.timerLiftIO[IO](global)
val task = IO.sleep(5.seconds) *> IO(println("Hello!"))

Observable.fromTaskLike(task)

Converts a Scala Try into an Observable.

import scala.util.Try

val value = Try(1)
Observable.fromTry(value)

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

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

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

Creates an Observable that emits auto-incremented natural numbers (longs) spaced by a given time interval. Starts from 0 with no delay, after which it emits incremented numbers spaced by the delay of time. The given delay of time acts as a fixed delay between successive events.

Creates an Observable that emits auto-incremented natural numbers (longs) at a fixed rate, as given by the specified period. The time it takes to process an onNext event gets subtracted from the specified period and thus the created observable tries to emit events spaced by the given time interval, regardless of how long the processing of onNext takes.

This version of the intervalAtFixedRate allows specifying an initialDelay before events start being emitted.

Creates an Observable that emits auto-incremented natural numbers (longs) at a fixed rate, as given by the specified period. The time it takes to process an onNext event gets subtracted from the specified period and thus the created observable tries to emit events spaced by the given time interval, regardless of how long the processing of onNext takes.

Creates an Observable that emits auto-incremented natural numbers (longs) spaced by a given time interval. Starts from 0 with no delay, after which it emits incremented numbers spaced by the delay of time. The given delay of time acts as a fixed delay between successive events.

Creates an Observable that emits auto-incremented natural numbers (longs) spaced by a given time interval. Starts from 0 with initialDelay, after which it emits incremented numbers spaced by the delay of time. The given delay of time acts as a fixed delay between successive events.

This version of the intervalWithFixedDelay allows specifying an initialDelay before events start being emitted.

Creates an input channel and an output observable pair for building a multicast data-source.

Useful for building multicast observables from data-sources that cannot be back-pressured.

Prefer Observable.create when possible.

Creates an input channel and an output observable pair for building a multicast data-source.

Useful for building multicast observables from data-sources that cannot be back-pressured.

Prefer Observable.create when possible.

Lifts an element into the Observable context.

Alias for now.

Creates an Observable that emits items in the given range.

Creates a Observable that depends on resource allocated by a monadic value, ensuring the resource is released.

Typical use-cases are working with files or network sockets

Example

import monix.eval.Task
import java.io.PrintWriter

val printer =
  Observable.resource {
    Task(new PrintWriter("./lines.txt"))
  } { writer =>
    Task(writer.close())
  }

// Safely use the resource, because the release is
// scheduled to happen afterwards
val writeLines = printer.flatMap { writer =>
  Observable
    .fromIterator(Task(Iterator.from(1)))
    .mapEval(i => Task { writer.println(s"Line #\$i") })
}

// Write 100 numbered lines to the file, closing the writer
// when finished (after `runAsync`):
writeLines.take(100).completedL

Creates a stream that depends on resource allocated by a monadic value, ensuring the resource is released.

Typical use-cases are working with files or network sockets

Example

import cats.effect.ExitCase
import monix.eval.Task
import java.io.PrintWriter

val printer =
  Observable.resourceCase {
    Task(new PrintWriter("./lines.txt"))
  } {
    case (writer, ExitCase.Canceled | ExitCase.Completed) =>
      Task(writer.close())
    case (writer, ExitCase.Error(e)) =>
      Task { println(e.getMessage); writer.close() }
  }

// Safely use the resource, because the release is
// scheduled to happen afterwards
val writeLines = printer.flatMap { writer =>
  Observable
    .fromIterator(Task(Iterator.from(1)))
    .mapEval(i => Task { writer.println(s"Line #\$i") })
}

// Write 100 numbered lines to the file, closing the writer
// when finished (after `runAsync`):
writeLines.take(100).completedL

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

So you can work among others with:

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

Version of resource 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
• ...

Keeps calling f and concatenating the resulting observables for each scala.util.Left event emitted by the source, concatenating the resulting observables and pushing every scala.util.Right[B] events downstream.

Based on Phil Freeman's Stack Safety for Free.

It helps to wrap your head around it if you think of it as being equivalent to this inefficient and unsafe implementation (for Observable):

// Don't do this kind of recursion, because `flatMap` can throw
// stack overflow errors:
def tailRecM[A, B](a: A)(f: (A) => Observable[Either[A, B]]): Observable[B] =
  f(a).flatMap {
    case Right(b) => Observable.pure(b)
    case Left(nextA) => tailRecM(nextA)(f)
 }

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

Given an initial state and a generator function that produces the next state and the next element in the sequence, creates an observable that keeps generating elements produced by our generator function until None is returned.

Given an initial state and a generator function that produces the next state and the next element in the sequence, creates an observable that keeps generating elements produced by our generator function until None is returned.

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

So you can work among others with:

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

Given a subscribe function, lifts it into an Observable.

This function is unsafe to use because users have to know and apply the Monix communication contract, related to thread-safety, communicating demand (back-pressure) and error handling.

Only use if you know what you're doing. Otherwise prefer create.

Creates a new observable from two observable sequences 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 items 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 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 combineLatestMap2 for a more relaxed alternative that doesn't combine items in strict sequence.

Creates a new observable from three observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from four observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from five observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from five observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from two observable sequences 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 items 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 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 combineLatestMap2 for a more relaxed alternative that doesn't combine items in strict sequence.

Creates a new observable from three observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from four observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from five observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

Creates a new observable from five observable sequences 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 items 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 items emitted by each of those observables; and so forth.

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

DEPRECATED — please switch to the concat method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).concat

DEPRECATED — please switch to the concatDelayErrors method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).concatDelayErrors

DEPRECATED — please switch to the flatten method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).flatten

DEPRECATED — please switch to the flattenDelayErrors method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).flattenDelayErrors

DEPRECATED — please use .executeOn.

The reason for the deprecation is the repurposing of the word "fork" in Task.

DEPRECATED — please use .executeAsync.

The reason for the deprecation is the repurposing of the word "fork" in Task.

DEPRECATED — switch to Observable.from.

DEPRECATED — switch to Observable.from.

DEPRECATED — please switch to the merge method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).merge

DEPRECATED — please switch to the merge method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).merge

DEPRECATED — please switch to the switch method.

This deprecation was made because there's no point in having this function described both as a method and as a companion object function. In general in API design we either have both for all functions, or we have to choose.

Switch to: Observable(list).switch