docs » cp.rx.Observable

Observables push values to Observers.

API Overview

Functions - API calls offered directly by the extension
is
Constructors - API calls which return an object, typically one that offers API methods
create
defer
empty
firstEmitting
fromCoroutine
fromFileByLine
fromRange
fromTable
never
of
replicate
throw
zip
Methods - API calls which can only be made on an object returned by a constructor
all
average
buffer
catch
combineLatest
compact
concat
contains
count
debounce
defaultIfEmpty
delay
distinct
distinctUntilChanged
dump
elementAt
filter
finalize
find
first
flatMap
flatMapLatest
flatten
ignoreElements
last
map
max
merge
min
next
partition
pluck
reduce
reject
retry
retryWithDelay
sample
scan
skip
skipLast
skipUntil
skipWhile
startWith
subscribe
sum
switch
switchIfEmpty
take
takeLast
takeUntil
takeWhile
tap
timeout
unpack
unwrap
with

API Documentation

Functions

is

Signature	`cp.rx.Observable.is(thing) -> boolean`
Type	Function
Description	Checks if the thing is an instance of Observable.
Parameters	thing - The thing to check.
Returns	`true` if the thing is an `Observable`.

Constructors

create

Signature	`cp.rx.Observable.create(onSubscription) -> cp.rx.Observable`
Type	Constructor
Description	Creates a new Observable.
Parameters	onSubscription - The reference function that produces values.
Returns	The new `Observable`.

defer

Signature	`cp.rx.Observable.defer(fn) -> cp.rx.Observable`
Type	Constructor
Description	Creates an `Observable` that executes the `function` to create a new `Observable` each time an Observer subscribes.
Parameters	fn - A function that returns an `Observable`.
Returns	The new `Observable`.

empty

Signature	`cp.rx.Observable.empty() -> cp.rx.Observable`
Type	Constructor
Description	Returns an Observable that immediately completes without producing a value.
Parameters	None
Returns	The new `Observable`.

firstEmitting

Signature	`cp.rx.Observable.firstEmitting(...) -> cp.rx.Observer`
Type	Constructor
Description	Given a set of Observables, produces values from only the first one to produce a value or complete.
Parameters	... - list of Observables
Returns	The new `Observable`.

fromCoroutine

Signature	`cp.rx.Observable.fromCoroutine(fn, scheduler) -> cp.rx.Observable`
Type	Constructor
Description	Creates an Observable that produces values when the specified coroutine yields.
Parameters	fn - A `coroutine` or `function` to use to generate values. Note that if a coroutine is used, the values it yields will be shared by all subscribed Observers (influenced by the Scheduler), whereas a new coroutine will be created for each Observer when a `function` is used.
Returns	The new `Observable`.

fromFileByLine

Signature	`cp.rx.Observable.fromFileByLine(filename) -> cp.rx.Observable`
Type	Constructor
Description	Creates an Observable that produces values from a file, line by line.
Parameters	filename - The name of the file used to create the Observable.
Returns	The new `Observable`.

fromRange

Signature	`cp.rx.Observable.fromRange(initial[, limit[, step]]) -> cp.rx.Observable`
Type	Constructor
Description	Creates an Observable that produces a range of values in a manner similar to a Lua `for` loop.
Parameters	initial - The first value of the range, or the upper limit if no other arguments are specified. limit - The second value of the range. Defaults to no limit. step - An amount to increment the value by each iteration. Defaults to `1`.
Returns	The new `Observable`.

fromTable

Signature	`cp.rx.Observable.fromTable(t, iterator, keys) -> cp.rx.Observable`
Type	Constructor
Description	Creates an `Observable` that produces values from a table.
Parameters	t - The `table` used to create the `Observable`. iterator - An iterator used to iterate the table, e.g. `pairs` or `ipairs`. Defaults to `pairs`. keys - If `true`, also emit the keys of the table. Defaults to `false`.
Returns	The new `Observable`.

never

Signature	`cp.rx.Observable.never() -> cp.rx.Observable`
Type	Constructor
Description	Returns an Observable that never produces values and never completes.
Parameters	None
Returns	The new `Observable`.

of

Signature	`cp.rx.Observable.of(...) -> cp.rx.Observable`
Type	Constructor
Description	Creates an Observable that produces a set of values.
Parameters	... - The list of values to send as individual `onNext` values.
Returns	The new `Observable`.

replicate

Signature	`cp.rx.Observable.replicate(value[, count]) -> cp.rx.Observable`
Type	Constructor
Description	Creates an `Observable` that repeats a value a specified number of times.
Parameters	value - The value to repeat. count - The number of times to repeat the value. If left unspecified, the value is repeated an infinite number of times.
Returns	The new `Observable`.

throw

Signature	`cp.rx.Observable.throw(message, ...) -> cp.rx.Observable`
Type	Constructor
Description	Returns an Observable that immediately produces an error.
Parameters	message - The message to send. ... - The additional values to apply to the message, using `string.format` syntax.
Returns	The new `Observable`.

zip

Signature	`cp.rx.Observable.zip(...) -> cp.rx.Observable`
Type	Constructor
Description	Returns an `Observable` that merges the values produced by the source `Observables` by grouping them
Parameters	... - The `Observables` to zip.
Returns	The new `Observable`.

Methods

all

Signature	`cp.rx.Observable:all(predicate) -> cp.rx.Observable`
Type	Method
Description	Determine whether all items emitted by an Observable meet some criteria.
Parameters	predicate - The predicate used to evaluate objects. Defaults to the `identity`.
Returns	A new `Observable`.

average

Signature	`cp.rx.Observable:average() -> cp.rx.Observable`
Type	Method
Description	Returns an Observable that produces the average of all values produced by the original.
Returns	The new `Observable`.

buffer

Signature	`cp.rx.Observable:buffer(size) -> cp.rx.Observable`
Type	Method
Description	Returns an Observable that buffers values from the original and produces them as multiple values.
Parameters	size - The size of the buffer.
Returns	The new `Observable`.

catch

Signature	`cp.rx.Observable:catch(handler) -> cp.rx.Observable`
Type	Method
Description	Returns an Observable that intercepts any errors from the previous and replace them with values
Parameters	handler - An `Observable` or a `function` that returns an `Observable` to replace the source `Observable` in the event of an error.
Returns	The new `Observable`.

combineLatest

Signature	`cp.rx.Observable:combineLatest(...) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that runs a combinator function on the most recent values from a set
Parameters	... - One or more `Observables` to combine. combinator - A `function` that combines the latest result from each `Observable` and returns a single value.
Returns	The new `Observable`.

compact

Signature	`cp.rx.Observable:compact() -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the values of the first with falsy values removed.
Returns	The new `Observable`.

concat

Signature	`cp.rx.Observable:concat(...) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the values produced by all the specified `Observables` in
Parameters	... - The list of `Observables` to concatenate.
Returns	The new `Observable`.

contains

Signature	`cp.rx.Observable:contains(value) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces a single `boolean` value representing whether or not the
Parameters	value - The value to search for. `==` is used for equality testing.
Returns	The new `Observable`.

count

Signature	`cp.rx.Observable:count([predicate]) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that produces a single value representing the number of values produced
Parameters	predicate - The predicate `function` used to match values.
Returns	The new `Observable`.

debounce

Signature	`cp.rx.Observable:debounce(time[, scheduler]) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that mirrors the source `Observable`, except that it drops items emitted by the source
Parameters	time - The number of milliseconds. scheduler - The scheduler. Uses the defaultScheduler by default.
Returns	The new `Observable`.

defaultIfEmpty

Signature	`cp.rx.Observable:defaultIfEmpty(...)`
Type	Method
Description	Returns a new `Observable` that produces a default set of items if the source `Observable` produces
Returns	The new `Observable`.

delay

Signature	`cp.rx.Observable:delay(time, scheduler) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the values of the original delayed by a time period.
Parameters	time - An amount in milliseconds to delay by, or a `function` which returns this value. scheduler - The Scheduler to run the `Observable` on.
Returns	The new `Observable`.

distinct

Signature	`cp.rx.Observable:distinct() -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the values from the original with duplicates removed.
Returns	The new `Observable`.

distinctUntilChanged

Signature	`cp.rx.Observable:distinctUntilChanged([comparator]) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that only produces values from the original if they are different from
Parameters	comparator - A `function` used to compare 2 values. If unspecified, `==` is used.
Returns	The new `Observable`

dump

Signature	`cp.rx.Observable:dump(name, formatter)`
Type	Method
Description	Subscribes to this Observable and prints values it produces.
Parameters	name - Prefixes the printed messages with a name. formatter - A function that formats one or more values to be printed. Defaults to `tostring`.
Returns	A Reference for the subscription.

elementAt

Signature	`cp.rx.Observable:elementAt(index) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that produces the `n`th element produced by the source `Observable`.
Parameters	index - The index of the item, with an index of `1` representing the first.
Returns	The new `Observable`.

filter

Signature	`cp.rx.Observable:filter(predicate) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that only produces values of the first that satisfy a predicate.
Parameters	predicate - The predicate `function` used to filter values.
Returns	The new `Observable`.

finalize

Signature	`cp.rx.Observable:finalize(handler) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that mirrors the source `Observable`, but will call a specified `function`
Parameters	handler - The handler `function` to call when `onError`/`onCompleted` occurs.
Returns	The new `Observable`.

find

Signature	`cp.rx.Observable:find(predicate) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the first value of the original that satisfies a
Parameters	predicate - The predicate `function` used to find a value.
Returns	The new `Observable`.

first

Signature	`cp.rx.Observable:first() -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that only produces the first result of the original.
Returns	The new `Observable`.

flatMap

Signature	`cp.rx.Observable:flatMap(callback) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that transform the items emitted by an `Observable` into `Observables`,
Parameters	callback - The `function` to transform values from the original `Observable`.
Returns	The new `Observable`.

flatMapLatest

Signature	`cp.rx.Observable:flatMapLatest([callback]) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that uses a callback to create `Observables` from the values produced by
Parameters	@arg {function=identity} callback - The function used to convert values to Observables. Defaults to the identity function.
Returns	The new `Observable`.

flatten

Signature	`cp.rx.Observable:flatten()`
Type	Method
Description	Returns a new `Observable` that subscribes to the `Observables` produced by the original and
Returns	The new `Observable`.

ignoreElements

Signature	`cp.rx.Observable:ignoreElements() -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that terminates when the source terminates but does not produce any
Returns	The new `Observable`.

last

Signature	`cp.rx.Observable:last() -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that only produces the last result of the original.
Returns	The new `Observable`.

map

Signature	`cp.rx.Observable:map(callback) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the values of the original transformed by a `function`.
Parameters	callback - The `function` to transform values from the original `Observable`.
Returns	The new `Observable`.

max

Signature	`cp.rx.Observable:max() -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the maximum value produced by the original.
Returns	The new `Observable`.

merge

Signature	`cp.rx.Observable:merge(...) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the values produced by all the specified `Observables` in
Parameters	... - One or more `Observables` to merge.
Returns	The new `Observable`.

min

Signature	`cp.rx.Observable:min() -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the minimum value produced by the original.
Returns	The new `Observable`.

partition

Signature	`cp.rx.Observable:partition(predicate) -> cp.rx.Observable, cp.rx.Observable`
Type	Method
Description	Returns two `Observables`: one that produces values for which the predicate returns truthy for,
Parameters	predicate - The predicate `function` used to partition the values.
Returns	The 'truthy' `Observable`. The 'falsy' `Observable`.

pluck

Signature	`cp.rx.Observable:pluck(...) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces values computed by extracting the given keys from the
Parameters	... - The key to extract from the `table`. Multiple keys can be specified to recursively pluck values from nested tables.
Returns	The new `Observable`.

reduce

Signature	`cp.rx.Observable:reduce(accumulator[, seed]) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces a single value computed by accumulating the results of
Parameters	accumulator - Accumulates the values of the original `Observable`. Will be passed the return value of the last call as the first argument and the current values as the rest of the arguments. seed - A value to pass to the accumulator the first time it is run.
Returns	The new `Observable`.

reject

Signature	`cp.rx.Observable:reject(predicate) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces values from the original which do not satisfy a predicate.
Parameters	predicate - The predicate `function` used to reject values.
Returns	The new `Observable`.

retry

Signature	`cp.rx.Observable:retry([count]) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that restarts in the event of an error.
Parameters	count - The maximum number of times to retry. If left unspecified, an infinite number of retries will be attempted.
Returns	The new `Observable`.

retryWithDelay

Signature	`cp.rx.Observable:retryWithDelay(count[, delay[, scheduler]]) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that restarts in the event of an error.
Parameters	count - The maximum number of times to retry. If left unspecified, an infinite number of retries will be attempted. delay - The `function` returning or a `number` representing the delay in milliseconds or a `function`. If left unspecified, defaults to 1000 ms (1 second). scheduler - The Scheduler to use. If not specified, it will use the [defaultScheduler](cp.rx.util#defaultScheduler].
Returns	The new `Observable`.

sample

Signature	`cp.rx.Observable:sample(sampler) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces its most recent value every time the specified observable
Parameters	sampler - The `Observable` that is used to sample values from this `Observable`.

scan

Signature	`cp.rx.Observable:scan(accumulator, seed) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces values computed by accumulating the results of running a
Parameters	accumulator - Accumulates the values of the original `Observable`. Will be passed the return value of the last call as the first argument and the current values as the rest of the arguments. Each value returned from this `function` will be emitted by the `Observable`. seed - A value to pass to the accumulator the first time it is run.
Returns	The new `Observable`.

skip

Signature	`cp.rx.Observable:skip([n]) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that skips over a specified number of values produced by the original
Parameters	n - The number of values to ignore. Defaults to `1`.
Returns	The new `Observable`

skipLast

Signature	`cp.rx.Observable:skipLast(count) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that omits a specified number of values from the end of the original `Observable`.
Parameters	count - The number of items to omit from the end.
Returns	The new `Observable`.

skipUntil

Signature	`cp.rx.Observable:skipUntil(other) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that skips over values produced by the original until the specified
Parameters	other - The `Observable` that triggers the production of values.
Returns	The new `Observable`.

skipWhile

Signature	`cp.rx.Observable:skipWhile(predicate) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that skips elements until the predicate returns `falsy` for one of them.
Parameters	predicate - The predicate `function` used to continue skipping values.
Returns	The new `Observable`.

startWith

Signature	`cp.rx.Observable:startWith(...) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces the specified values followed by all elements produced by
Parameters	values - The values to produce before the Observable begins producing values normally.
Returns	The new `Observable`.

| Signature | cp.rx.Observable:subscribe(observer | onNext[, onError[, onCompleted]]) -> cp.rx.Reference | | -----------------------------------------------------|---------------------------------------------------------------------------------------------------------| | Type | Method | | Description | Shorthand for creating an Observer and passing it to this Observable's subscription function. | | Parameters |

observer | onNext - Either an Observer or a function to be called when the Observable produces a value.
onError - A function to be called when the Observable terminates due to an error.
onCompleted - A 'function` to be called when the Observable completes normally.

| | Returns |

A Reference which can be used to cancel the subscription.

|

sum

Signature	`cp.rx.Observable:sum() -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that produces a single value representing the sum of the values produced
Returns	The new `Observable`.

switch

Signature	`cp.rx.Observable:switch() -> cp.rx.Observable`
Type	Method
Description	Given an `Observable` that produces `Observables`, returns an `Observable` that produces the values
Returns	The new `Observable`.

switchIfEmpty

Signature	`cp.rx.Observable:switchIfEmpty(alternate) -> cp.rx.Observable`
Type	Method
Description	Switch to an alternate `Observable` if this one sends an `onCompleted` without any `onNext`s.
Parameters	alternate - An `Observable` to switch to if this does not send any `onNext` values before the `onCompleted`.
Returns	The new `Observable`.

take

Signature	`cp.rx.Observable:take([n]) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that only produces the first n results of the original.
Parameters	n - The number of elements to produce before completing. Defaults to `1`.
Returns	The new `Observable`.

takeLast

Signature	`cp.rx.Observable:takeLast(count) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that produces a specified number of elements from the end of a source
Parameters	count - The number of elements to produce.
Returns	The new `Observable`.

takeUntil

Signature	`cp.rx.Observable:takeUntil(other) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that completes when the specified `Observable` fires.
Parameters	other - The `Observable` that triggers completion of the original.
Returns	The new `Observable`.

takeWhile

Signature	`cp.rx.Observable:takeWhile(predicate) -> cp.rx.Observable`
Type	Method
Description	Returns a new `Observable` that produces elements until the predicate returns `falsy`.
Parameters	predicate - The predicate `function` used to continue production of values.
Returns	The new `Observable`.

tap

Signature	`cp.rx.Observable:tap(onNext[, onError[, onCompleted]]) -> cp.rx.Observable`
Type	Method
Description	Runs a `function` each time this `Observable` has activity. Similar to subscribe but does not
Parameters	onNext - Run when the `Observable` produces values. onError - Run when the `Observable` encounters a problem. onCompleted - Run when the `Observable` completes.
Returns	The new `Observable`.

timeout

Signature	`cp.rx.Observable:timeout(timeInMs, next[, scheduler]) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that will emit an error if the specified time is exceded since the most recent `next` value.
Parameters	timeInMs - The time in milliseconds to wait before an error is emitted. next - If a `string`, it will be sent as an error. If an `Observable`, switch to that `Observable` instead of sending an error. scheduler - The scheduler to use. Uses the defaultScheduler if not provided.
Returns	The new `Observable`.

unpack

Signature	`cp.rx.Observable:unpack() -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that unpacks the `tables` produced by the original.
Returns	The new `Observable`.

unwrap

Signature	`cp.rx.Observable:unwrap() -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that takes any values produced by the original that consist of multiple
Returns	The new `Observable`.

with

Signature	`cp.rx.Observable:with(...) -> cp.rx.Observable`
Type	Method
Description	Returns an `Observable` that produces values from the original along with the most recently
Parameters	... - The `Observables` to include the most recent values from.
Returns	The new `Observable`.

Files

cp.rx.Observable.md

Latest commit

History