U
vij @ s$ d dl mZ d dlmZmZmZmZmZmZm Z m
Z
m Z m Z m
Z
mZmZmZ d dlmZmZmZmZmZmZmZ d dlmZ d dlmZ d dl m Z d dl m Z m Z m!Z!m"Z"m#Z#m$Z$ e
dZ%e
d Z&e
d
Z'e
d Z(e
d Z)e
d
Z*e
dZ+e
dZ,e
dZ-e
dZ.e
dZ/e
dZ0e#e% eee% gee1 f dddZ2ee% eee% gee% f dddZ3eee% gee% f ddd Z4de e!e%e5f eee% gee5 f d d d Z6ee eee% geee% f d!d"d#Z7eg ee f eee% geee% f d$d%d&Z8ee eegee f eee% geee% f d'd(d)Z9de:e e: eee% geee% f d*d+d,Z;dejdejZDde%e ej e% eee% gee1 f d?d@dAZEde ej#e% eee% gee: f ddBdCZFdejZde e!e&ee' f eee& gee' f d<d?d@Zde e"e&ee' f eee& gee' f d<dAdBZe:eee% gee% f ddCdDZe:eee% gee% f ddEdFZe:eee% geee% f ddGdHZdejdDdFdHdJdLdOdRdTdVdYd\dadddgdkdndqdsddwdyd{d}ddddddddddgZd S ( )Future)
TYPE_CHECKINGAnyCallableDictIterableListOptionalSetTupleTypeTypeVarUnioncastoverload)ConnectableObservableGroupedObservable Notification
Observableabccomposetyping)identity)NotSet)Subject) AccumulatorComparerMapper
MapperIndexed PredicatePredicateIndexed_T_T1_T2_TKey_TState_TValue_TRight_TLeft_A_B_C_D) predicatereturnc C s ddl m} || S )ak Determines whether all elements of an observable sequence satisfy
a condition.
.. marble::
:alt: all
--1--2--3--4--5-|
[ all(i: i<10) ]
----------------true-|
Example:
>>> op = all(lambda value: value.length > 3)
Args:
predicate: A function to test each element for a condition.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element
determining whether all elements in the source sequence pass
the test in the specified predicate.
)all_)Z_allr0 )r- r0 r1 M/opt/alt/python38/lib/python3.8/site-packages/reactivex/operators/__init__.pyall: s r3 ) right_sourcer. c C s ddl m} || S )a Propagates the observable sequence that reacts first.
.. marble::
:alt: amb
---8--6--9-----------|
--1--2--3---5--------|
----------10-20-30---|
[ amb() ]
--1--2--3---5--------|
Example:
>>> op = amb(ys)
Returns:
An operator function that takes an observable source and
returns an observable sequence that surfaces any of the given
sequences, whichever reacted first.
r/ )amb_)Z_ambr5 )r4 r5 r1 r1 r2 ambV s r6 )r. c C s ddl m} | S )zHides the identity of an observable sequence.
Returns:
An operator function that takes an observable source and
returns and observable sequence that hides the identity of the
source sequence.
r/ as_observable_)Z
_asobservabler8 r7 r1 r1 r2
as_observableo s r9 N)
key_mapperr. c C s ddl m} || S )a The average operator.
Computes the average of an observable sequence of values that
are in the sequence or obtained by invoking a transform function on
each element of the input sequence if present.
.. marble::
:alt: average
---1--2--3--4----|
[ average() ]
-----------------2.5-|
Examples:
>>> op = average()
>>> op = average(lambda x: x.value)
Args:
key_mapper: [Optional] A transform function to apply to each element.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element with
the average of the sequence of values.
r/ )average_)Z_averager; )r: r; r1 r1 r2 average| s r< )
boundariesr. c C s ddl m} || S )a8 Projects each element of an observable sequence into zero or
more buffers.
.. marble::
:alt: buffer
---a-----b-----c--------|
--1--2--3--4--5--6--7---|
[ buffer() ]
---1-----2,3---4,5------|
Examples:
>>> res = buffer(reactivex.interval(1.0))
Args:
boundaries: Observable sequence whose elements denote the
creation and completion of buffers.
Returns:
A function that takes an observable source and returns an
observable sequence of buffers.
r/ )buffer_)_bufferr> )r= r> r1 r1 r2 buffer s r@ )closing_mapperr. c C s ddl m} || S )a6 Projects each element of an observable sequence into zero or
more buffers.
.. marble::
:alt: buffer_when
--------c-|
--------c-|
--------c-|
---1--2--3--4--5--6-------|
[ buffer_when() ]
+-------1,2-----3,4,5---6-|
Examples:
>>> res = buffer_when(lambda: reactivex.timer(0.5))
Args:
closing_mapper: A function invoked to define the closing of each
produced buffer. A buffer is started when the previous one is
closed, resulting in non-overlapping buffers. The buffer is closed
when one item is emitted or when the observable completes.
Returns:
A function that takes an observable source and returns an
observable sequence of windows.
r/ ) buffer_when_)r? rB )rA rB r1 r1 r2 buffer_when s rC )openingsrA r. c C s ddl m} || |S )a Projects each element of an observable sequence into zero or
more buffers.
.. marble::
:alt: buffer_toggle
---a-----------b--------------|
---d--|
--------e--|
----1--2--3--4--5--6--7--8----|
[ buffer_toggle() ]
------1----------------5,6,7--|
>>> res = buffer_toggle(reactivex.interval(0.5), lambda i: reactivex.timer(i))
Args:
openings: Observable sequence whose elements denote the
creation of buffers.
closing_mapper: A function invoked to define the closing of each
produced buffer. Value from openings Observable that initiated
the associated buffer is provided as argument to the function. The
buffer is closed when one item is emitted or when the observable
completes.
Returns:
A function that takes an observable source and returns an
observable sequence of windows.
r/ )buffer_toggle_)r? rE )rD rA rE r1 r1 r2
buffer_toggle s rF )countskipr. c C s ddl m} || |S )a Projects each element of an observable sequence into zero or more
buffers which are produced based on element count information.
.. marble::
:alt: buffer_with_count
----1-2-3-4-5-6------|
[buffer_with_count(3)]
--------1,2,3-4,5,6--|
Examples:
>>> res = buffer_with_count(10)(xs)
>>> res = buffer_with_count(10, 1)(xs)
Args:
count: Length of each buffer.
skip: [Optional] Number of elements to skip between
creation of consecutive buffers. If not provided, defaults to
the count.
Returns:
A function that takes an observable source and returns an
observable sequence of buffers.
r/ )buffer_with_count_)r? rI )rG rH rI r1 r1 r2 buffer_with_count s rJ )timespan timeshift schedulerr. c C s ddl m} || ||S )a Projects each element of an observable sequence into zero or more
buffers which are produced based on timing information.
.. marble::
:alt: buffer_with_time
---1-2-3-4-5-6-----|
[buffer_with_time()]
-------1,2,3-4,5,6-|
Examples:
>>> # non-overlapping segments of 1 second
>>> res = buffer_with_time(1.0)
>>> # segments of 1 second with time shift 0.5 seconds
>>> res = buffer_with_time(1.0, 0.5)
Args:
timespan: Length of each buffer (specified as a float denoting seconds
or an instance of timedelta).
timeshift: [Optional] Interval between creation of consecutive buffers
(specified as a float denoting seconds or an instance of timedelta).
If not specified, the timeshift will be the same as the timespan
argument, resulting in non-overlapping adjacent buffers.
scheduler: [Optional] Scheduler to run the timer on. If not specified,
the timeout scheduler is used
Returns:
An operator function that takes an observable source and
returns an observable sequence of buffers.
r/ )buffer_with_time_)Z_bufferwithtimerN )rK rL rM rN r1 r1 r2 buffer_with_time! s # rO )rK rG rM r. c C s ddl m} || ||S )a Projects each element of an observable sequence into a buffer
that is completed when either it's full or a given amount of time
has elapsed.
.. marble::
:alt: buffer_with_time_or_count
--1-2-3-4-5-6------|
[ buffer() ]
------1,2,3-4,5,6--|
Examples:
>>> # 5s or 50 items in an array
>>> res = source._buffer_with_time_or_count(5.0, 50)
>>> # 5s or 50 items in an array
>>> res = source._buffer_with_time_or_count(5.0, 50, Scheduler.timeout)
Args:
timespan: Maximum time length of a buffer.
count: Maximum element count of a buffer.
scheduler: [Optional] Scheduler to run buffering timers on. If
not specified, the timeout scheduler is used.
Returns:
An operator function that takes an observable source and
returns an observable sequence of buffers.
r/ )buffer_with_time_or_count_)Z_bufferwithtimeorcountrP )rK rG rM rP r1 r1 r2 buffer_with_time_or_countI s rQ )handlerr. c C s ddl m} || S )a Continues an observable sequence that is terminated by an
exception with the next observable sequence.
.. marble::
:alt: catch
---1---2---3-*
a-7-8-|
[ catch(a) ]
---1---2---3---7-8-|
Examples:
>>> op = catch(ys)
>>> op = catch(lambda ex, src: ys(ex))
Args:
handler: Second observable sequence used to produce
results when an error occurred in the first sequence, or an
exception handler function that returns an observable sequence
given the error and source observable that occurred in the
first sequence.
Returns:
A function taking an observable source and returns an
observable sequence containing the first sequence's elements,
followed by the elements of the handler sequence in case an
exception occurred.
r/ )catch_)Z_catchrS )rR rS r1 r1 r2 catchn s ! rT )othersr. c G s ddl m} || S )az Merges the specified observable sequences into one observable
sequence by creating a tuple whenever any of the
observable sequences produces an element.
.. marble::
:alt: combine_latest
---a-----b--c------|
--1---2--------3---|
[ combine_latest() ]
---a1-a2-b2-c2-c3--|
Examples:
>>> obs = combine_latest(other)
>>> obs = combine_latest(obs1, obs2, obs3)
Returns:
An operator function that takes an observable sources and
returns an observable sequence containing the result of
combining elements of the sources into a tuple.
r/ )combine_latest_)Z_combinelatestrV )rU rV r1 r1 r2 combine_latest s rW )sourcesr. c G s ddl m} || S )a Concatenates all the observable sequences.
.. marble::
:alt: concat
---1--2--3--|
--6--8--|
[ concat() ]
---1--2--3----6--8-|
Examples:
>>> op = concat(xs, ys, zs)
Returns:
An operator function that takes one or more observable sources and
returns an observable sequence that contains the elements of
each given sequence, in sequential order.
r/ )concat_)Z_concatrY )rX rY r1 r1 r2 concat s rZ )projectr. c C s t t| tddS )a; Projects each source value to an Observable which is merged in the
output Observable, in a serialized fashion waiting for each one to complete
before merging the next.
Warning: if source values arrive endlessly and faster than their corresponding
inner Observables can complete, it will result in memory issues as inner
Observables amass in an unbounded buffer waiting
for their turn to be subscribed to.
Note: concatMap is equivalent to mergeMap with concurrency parameter set to 1.
.. marble::
:alt: concat_map
---1------------2-----3----------------------------|
[ concat_map(i: 10*i---10*i---10*i|) ]
---10---10---10-20---20---(20,30)---30---30--------|
Examples:
>>> op = concat(lambda i: reactivex.timer(1.0).pipe(take(3)))
Args:
project: Projecting function which takes the outer observable value
and emits the inner observable
Returns:
An operator function that maps each value to the inner observable
and emits its values in order.
r/ max_concurrent)r mapmerger[ r1 r1 r2
concat_map s "ra )valuecomparerr. c C s ddl m} || |S )a Determines whether an observable sequence contains a specified
element with an optional equality comparer.
.. marble::
:alt: contains
--1--2--3--4--|
[ contains(3) ]
--------------true-|
Examples:
>>> op = contains(42)
>>> op = contains({ "value": 42 }, lambda x, y: x["value"] == y["value"])
Args:
value: The value to locate in the source sequence.
comparer: [Optional] An equality comparer to compare elements.
Returns:
A function that takes a source observable that returns an
observable sequence containing a single element determining
whether the source sequence contains an element that has the
specified value.
r/ ) contains_)Z _containsrd )rb rc rd r1 r1 r2 contains s re c C s ddl m} || S )a) Returns an observable sequence containing a value that
represents how many elements in the specified observable sequence
satisfy a condition if provided, else the count of items.
.. marble::
:alt: count
--1--2--3--4--|
[ count(i: i>2) ]
--------------2-|
Examples:
>>> op = count()
>>> op = count(lambda x: x > 3)
Args:
predicate: A function to test each element for a condition.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element with
a number that represents how many elements in the input
sequence satisfy the condition in the predicate function if
provided, else the count of items in the sequence.
r/ )count_)_countrf )r- rf r1 r1 r2 rG s rG )duetimerM r. c C s ddl m} || |S )at Ignores values from an observable sequence which are followed by
another value before duetime.
.. marble::
:alt: debounce
--1--2-3-4--5------|
[ debounce() ]
----1------4---5---|
Example:
>>> res = debounce(5.0) # 5 seconds
Args:
duetime: Duration of the throttle period for each value
(specified as a float denoting seconds or an instance of timedelta).
scheduler: Scheduler to debounce values on.
Returns:
An operator function that takes the source observable and
returns the debounced observable sequence.
r/ ) debounce_) _debounceri )rh rM ri r1 r1 r2 debounce0 s rk )
default_valuer. c C s d S Nr1 rl r1 r1 r2 default_if_emptyQ s ro c C s d S rm r1 r1 r1 r1 r2 ro W s c C s ddl m} || S )a Returns the elements of the specified sequence or the specified
value in a singleton sequence if the sequence is empty.
.. marble::
:alt: default_if_empty
----------|
[default_if_empty(42)]
----------42-|
Examples:
>>> res = obs = default_if_empty()
>>> obs = default_if_empty(False)
Args:
default_value: The value to return if the sequence is empty. If
not provided, this defaults to None.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the specified
default value if the source is empty otherwise, the elements of
the source.
r/ )default_if_empty_)Z_defaultifemptyrp )rl rp r1 r1 r2 ro [ s c C s ddl m} || |dS )a Time shifts the observable sequence by delaying the
subscription.
.. marble::
:alt: delay_subscription
----1--2--3--4-----|
[ delay() ]
--------1--2--3--4-|
Example:
>>> res = delay_subscription(5.0) # 5s
Args:
duetime: Absolute or relative time to perform the subscription
at.
scheduler: Scheduler to delay subscription on.
Returns:
A function that take a source observable and returns a
time-shifted observable sequence.
r/ )delay_subscription_rM )Z_delaysubscriptionrq )rh rM rq r1 r1 r2 delay_subscription{ s rs )subscription_delaydelay_duration_mapperr. c C s ddl m} || |S )a Time shifts the observable sequence based on a subscription
delay and a delay mapper function for each element.
.. marble::
:alt: delay_with_mapper
----1--2--3--4-----|
[ delay() ]
--------1--2--3--4-|
Examples:
>>> # with mapper only
>>> res = source.delay_with_mapper(lambda x: reactivex.timer(5.0))
>>> # with delay and mapper
>>> res = source.delay_with_mapper(
reactivex.timer(2.0), lambda x: reactivex.timer(x)
)
Args:
subscription_delay: [Optional] Sequence indicating the delay
for the subscription to the source.
delay_duration_mapper: [Optional] Selector function to retrieve
a sequence indicating the delay for each given element.
Returns:
A function that takes an observable source and returns a
time-shifted observable sequence.
r/ )delay_with_mapper_)Z_delaywithmapperrv )rt ru rv r1 r1 r2 delay_with_mapper s $ rw c C s ddl m} | S )a Dematerialize operator.
Dematerializes the explicit notification values of an
observable sequence as implicit notifications.
Returns:
An observable sequence exhibiting the behavior
corresponding to the source sequence's notification values.
r/ dematerialize_)Z_dematerializery rx r1 r1 r2
dematerialize s
rz c C s ddl m} || |S )a The delay operator.
.. marble::
:alt: delay
----1--2--3--4-----|
[ delay() ]
--------1--2--3--4-|
Time shifts the observable sequence by duetime. The relative time
intervals between the values are preserved.
Examples:
>>> res = delay(timedelta(seconds=10))
>>> res = delay(5.0)
Args:
duetime: Relative time, specified as a float denoting seconds or an
instance of timedelta, by which to shift the observable sequence.
scheduler: [Optional] Scheduler to run the delay timers on.
If not specified, the timeout scheduler is used.
Returns:
A partially applied operator function that takes the source
observable and returns a time-shifted sequence.
r/ )delay_)Z_delayr{ )rh rM r{ r1 r1 r2 delay s r| )r: rc r. c C s ddl m} || |S )a Returns an observable sequence that contains only distinct
elements according to the key_mapper and the comparer. Usage of
this operator should be considered carefully due to the maintenance
of an internal lookup structure which can grow large.
.. marble::
:alt: distinct
-0-1-2-1-3-4-2-0---|
[ distinct() ]
-0-1-2---3-4-------|
Examples:
>>> res = obs = xs.distinct()
>>> obs = xs.distinct(lambda x: x.id)
>>> obs = xs.distinct(lambda x: x.id, lambda a,b: a == b)
Args:
key_mapper: [Optional] A function to compute the comparison
key for each element.
comparer: [Optional] Used to compare items in the collection.
Returns:
An operator function that takes an observable source and
returns an observable sequence only containing the distinct
elements, based on a computed key value, from the source
sequence.
r/ ) distinct_)Z _distinctr} )r: rc r} r1 r1 r2 distinct s ! r~ c C s ddl m} || |S )a Returns an observable sequence that contains only distinct
contiguous elements according to the key_mapper and the comparer.
.. marble::
:alt: distinct_until_changed
-0-1-1-2-3-1-2-2-3-|
[ distinct() ]
-0-1---2-3-1-2---3-|
Examples:
>>> op = distinct_until_changed();
>>> op = distinct_until_changed(lambda x: x.id)
>>> op = distinct_until_changed(lambda x: x.id, lambda x, y: x == y)
Args:
key_mapper: [Optional] A function to compute the comparison key
for each element. If not provided, it projects the value.
comparer: [Optional] Equality comparer for computed key values.
If not provided, defaults to an equality comparer function.
Returns:
An operator function that takes an observable source and
returns an observable sequence only containing the distinct
contiguous elements, based on a computed key value, from the
source sequence.
r/ )distinct_until_changed_)Z_distinctuntilchangedr )r: rc r r1 r1 r2 distinct_until_changed s r )observerr. c C s ddl m} || S )a Invokes an action for each element in the observable sequence
and invokes an action on graceful or exceptional termination of the
observable sequence. This method can be used for debugging,
logging, etc. of query behavior by intercepting the message stream
to run arbitrary actions for messages on the pipeline.
.. marble::
:alt: do
----1---2---3---4---|
[ do(i: foo()) ]
----1---2---3---4---|
>>> do(observer)
Args:
observer: Observer
Returns:
An operator function that takes the source observable and
returns the source sequence with the side-effecting behavior
applied.
r/ )do_)_dor )r r r1 r1 r2 do? s r )on_nexton_error on_completedr. c C s ddl m} || ||S )aT Invokes an action for each element in the observable sequence
and invokes an action on graceful or exceptional termination of the
observable sequence. This method can be used for debugging,
logging, etc. of query behavior by intercepting the message stream
to run arbitrary actions for messages on the pipeline.
.. marble::
:alt: do_action
----1---2---3---4---|
[do_action(i: foo())]
----1---2---3---4---|
Examples:
>>> do_action(send)
>>> do_action(on_next, on_error)
>>> do_action(on_next, on_error, on_completed)
Args:
on_next: [Optional] Action to invoke for each element in the
observable sequence.
on_error: [Optional] Action to invoke on exceptional
termination of the observable sequence.
on_completed: [Optional] Action to invoke on graceful
termination of the observable sequence.
Returns:
An operator function that takes the source observable an
returns the source sequence with the side-effecting behavior
applied.
r/ )
do_action_)r r )r r r r r1 r1 r2 do_action] s $ r ) conditionr. c C s ddl m} || S )a Repeats source as long as condition holds emulating a do while
loop.
.. marble::
:alt: do_while
--1--2--|
[ do_while() ]
--1--2--1--2--1--2--|
Args:
condition: The condition which determines if the source will be
repeated.
Returns:
An observable sequence which is repeated as long
as the condition holds.
r/ ) do_while_)Z_dowhiler )r r r1 r1 r2 do_while s r )indexr. c C s ddl m} || dS )a Returns the element at a specified index in a sequence.
.. marble::
:alt: element_at
----1---2---3---4---|
[ element_at(2) ]
------------3-|
Example:
>>> res = source.element_at(5)
Args:
index: The zero-based index of the element to retrieve.
Returns:
An operator function that takes an observable source and
returns an observable sequence that produces the element at
the specified position in the source sequence.
r/ element_at_or_default_FZ_elementatordefaultr )r r r1 r1 r2
element_at s r )r rl r. c C s ddl m} || d|S )aK Returns the element at a specified index in a sequence or a
default value if the index is out of range.
.. marble::
:alt: element_at_or_default
--1---2---3---4-|
[ element_at(6, a) ]
----------------a-|
Example:
>>> res = source.element_at_or_default(5)
>>> res = source.element_at_or_default(5, 0)
Args:
index: The zero-based index of the element to retrieve.
default_value: [Optional] The default value if the index is
outside the bounds of the source sequence.
Returns:
A function that takes an observable source and returns an
observable sequence that produces the element at the
specified position in the source sequence, or a default value if
the index is outside the bounds of the source sequence.
r/ r Tr )r rl r r1 r1 r2 element_at_or_default s r c C s ddl m} | S )a Performs a exclusive waiting for the first to finish before
subscribing to another observable. Observables that come in between
subscriptions will be dropped on the floor.
.. marble::
:alt: exclusive
-+---+-----+-------|
+-7-8-9-|
+-4-5-6-|
+-1-2-3-|
[ exclusive() ]
---1-2-3-----7-8-9-|
Returns:
An exclusive observable with only the results that
happen when subscribed.
r/
exclusive_)Z
_exclusiver r r1 r1 r2 exclusive s r )mapperr. c C s ddl m} || S )ar Expands an observable sequence by recursively invoking mapper.
Args:
mapper: Mapper function to invoke for each produced element,
resulting in another sequence to which the mapper will be
invoked recursively again.
Returns:
An observable sequence containing all the elements produced
by the recursive expansion.
r/ )expand_)_expandr )r r r1 r1 r2 expand s r c C s ddl m} || S )a9 Filters the elements of an observable sequence based on a
predicate.
.. marble::
:alt: filter
----1---2---3---4---|
[ filter(i: i>2) ]
------------3---4---|
Example:
>>> op = filter(lambda value: value < 10)
Args:
predicate: A function to test each source element for a
condition.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains elements from the
input sequence that satisfy the condition.
r/ )filter_)_filterr )r- r r1 r1 r2 filter s r )predicate_indexedr. c C s ddl m} || S )a Filters the elements of an observable sequence based on a
predicate by incorporating the element's index.
.. marble::
:alt: filter_indexed
----1---2---3---4---|
[ filter(i,id: id>2)]
----------------4---|
Example:
>>> op = filter_indexed(lambda value, index: (value + index) < 10)
Args:
predicate: A function to test each source element for a
condition; the second parameter of the function represents
the index of the source element.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains elements from the
input sequence that satisfy the condition.
r/ )filter_indexed_)r r )r r r1 r1 r2 filter_indexed$ s r )actionr. c C s ddl m} || S )av Invokes a specified action after the source observable sequence
terminates gracefully or exceptionally.
.. marble::
:alt: finally_action
--1--2--3--4--|
a-6-7-|
[finally_action(a)]
--1--2--3--4--6-7-|
Example:
>>> res = finally_action(lambda: print('sequence ended')
Args:
action: Action to invoke after the source observable sequence
terminates.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the action-invoking
termination behavior applied.
r/ )finally_action_)Z_finallyactionr )r r r1 r1 r2 finally_actionC s r c C s6 ddl m} tttt gtttdf f || dS )a Searches for an element that matches the conditions defined by
the specified predicate, and returns the first occurrence within
the entire Observable sequence.
.. marble::
:alt: find
--1--2--3--4--3--2--|
[ find(3) ]
--------3-|
Args:
predicate: The predicate that defines the conditions of the
element to search for.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the first element that
matches the conditions defined by the specified predicate, if
found otherwise, None.
r/ find_value_NF)_findr r r r r! r r- r r1 r1 r2 find` s
r c C s6 ddl m} tttt gtttdf f || dS )a Searches for an element that matches the conditions defined by
the specified predicate, and returns an Observable sequence with the
zero-based index of the first occurrence within the entire
Observable sequence.
.. marble::
:alt: find_index
--1--2--3--4--3--2--|
[ find_index(3) ]
--------2-|
Args:
predicate: The predicate that defines the conditions of the
element to search for.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the zero-based index of the
first occurrence of an element that matches the conditions
defined by match, if found; otherwise, -1.
r/ r NT)r r r r r r! r intr r1 r1 r2
find_index s
r c C s ddl m} || S )a Returns the first element of an observable sequence that
satisfies the condition in the predicate if present else the first
item in the sequence.
.. marble::
:alt: first
---1---2---3---4----|
[ first(i: i>1) ]
-------2-|
Examples:
>>> res = res = first()
>>> res = res = first(lambda x: x > 3)
Args:
predicate: [Optional] A predicate function to evaluate for
elements in the source sequence.
Returns:
A function that takes an observable source and returns an
observable sequence containing the first element in the
observable sequence that satisfies the condition in the
predicate if provided, else the first item in the sequence.
r/ )first_)Z_firstr )r- r r1 r1 r2 first s r )r- rl r. c C s ddl m} || |S )a Returns the first element of an observable sequence that
satisfies the condition in the predicate, or a default value if no
such element exists.
.. marble::
:alt: first_or_default
--1--2--3--4-|
[first(i: i>10, 42)]
-------------42-|
Examples:
>>> res = first_or_default()
>>> res = first_or_default(lambda x: x > 3)
>>> res = first_or_default(lambda x: x > 3, 0)
>>> res = first_or_default(None, 0)
Args:
predicate: [optional] A predicate function to evaluate for
elements in the source sequence.
default_value: [Optional] The default value if no such element
exists. If not specified, defaults to None.
Returns:
A function that takes an observable source and returns an
observable sequence containing the first element in the
observable sequence that satisfies the condition in the
predicate, or a default value if no such element exists.
r/ )first_or_default_)Z_firstordefaultr )r- rl r r1 r1 r2 first_or_default s r c C s d S rm r1 r r1 r1 r2 flat_map s r c C s d S rm r1 r r1 r1 r2 r s c C s d S rm r1 r r1 r1 r2 r s c C s d S rm r1 r r1 r1 r2 r s c C s ddl m} || S )a. The flat_map operator.
.. marble::
:alt: flat_map
--1-2-3-|
[ flat_map(range) ]
--0-0-1-0-1-2-|
One of the Following:
Projects each element of an observable sequence to an observable
sequence and merges the resulting observable sequences into one
observable sequence.
Example:
>>> flat_map(lambda x: Observable.range(0, x))
Or:
Projects each element of the source observable sequence to the
other observable sequence and merges the resulting observable
sequences into one observable sequence.
Example:
>>> flat_map(Observable.of(1, 2, 3))
Args:
mapper: A transform function to apply to each element or an
observable sequence to project each element from the source
sequence onto.
Returns:
An operator function that takes a source observable and returns
an observable sequence whose elements are the result of
invoking the one-to-many transform function on each element of
the input sequence.
r/ ) flat_map_)_flatmapr )r r r1 r1 r2 r s ( )mapper_indexedr. c C s d S rm r1 r r1 r1 r2 flat_map_indexed- s r c C s d S rm r1 r r1 r1 r2 r 3 s c C s d S rm r1 r r1 r1 r2 r 9 s c C s d S rm r1 r r1 r1 r2 r ? s c C s ddl m} || S )at The `flat_map_indexed` operator.
One of the Following:
Projects each element of an observable sequence to an observable
sequence and merges the resulting observable sequences into one
observable sequence.
.. marble::
:alt: flat_map_indexed
--1-2-3-|
[ flat_map(range) ]
--0-0-1-0-1-2-|
Example:
>>> source.flat_map_indexed(lambda x, i: Observable.range(0, x))
Or:
Projects each element of the source observable sequence to the
other observable sequence and merges the resulting observable
sequences into one observable sequence.
Example:
>>> source.flat_map_indexed(Observable.of(1, 2, 3))
Args:
mapper_indexed: [Optional] A transform function to apply to
each element or an observable sequence to project each
element from the source sequence onto.
Returns:
An operator function that takes an observable source and
returns an observable sequence whose elements are the result of
invoking the one-to-many transform function on each element of
the input sequence.
r/ )flat_map_indexed_)r r )r r r1 r1 r2 r E s ' z Future[_T2]c C s ddl m} || S )a Projects each element of an observable sequence into a new
sequence of observable sequences by incorporating the element's
index and then transforms an observable sequence of observable
sequences into an observable sequence producing values only from
the most recent observable sequence.
Args:
mapper: A transform function to apply to each source element.
The second parameter of the function represents the index
of the source element.
Returns:
An operator function that takes an observable source and
returns an observable sequence whose elements are the result of
invoking the transform function on each element of source
producing an observable of Observable sequences and that at any
point in time produces the elements of the most recent inner
observable sequence that has been received.
r/ )flat_map_latest_)r r )r r r1 r1 r2 flat_map_latestq s r .c G s ddl m} || S )a Wait for observables to complete and then combine last values
they emitted into a tuple. Whenever any of that observables completes
without emitting any value, result sequence will complete at that moment as well.
.. marble::
:alt: fork_join
---a-----b--c---d-|
--1---2------3-4---|
-a---------b---|
[ fork_join() ]
--------------------d4b|
Examples:
>>> res = fork_join(obs1)
>>> res = fork_join(obs1, obs2, obs3)
Returns:
An operator function that takes an observable source and
return an observable sequence containing the result
of combining last element from each source in given sequence.
r/ )
fork_join_)Z _forkjoinr )rU r r1 r1 r2 fork_join s r )r: element_mappersubject_mapperr. c C s ddl m} || ||S )a Groups the elements of an observable sequence according to a
specified key mapper function and comparer and selects the
resulting elements by using a specified function.
.. marble::
:alt: group_by
--1--2--a--3--b--c-|
[ group_by() ]
-+-----+-----------|
+a-----b--c-|
+1--2-----3-------|
Examples:
>>> group_by(lambda x: x.id)
>>> group_by(lambda x: x.id, lambda x: x.name)
>>> group_by(lambda x: x.id, lambda x: x.name, lambda: ReplaySubject())
Keyword arguments:
key_mapper: A function to extract the key for each element.
element_mapper: [Optional] A function to map each source
element to an element in an observable group.
subject_mapper: A function that returns a subject used to initiate
a grouped observable. Default mapper returns a Subject object.
Returns:
An operator function that takes an observable source and
returns a sequence of observable groups, each of which
corresponds to a unique key value, containing all elements that
share that same key value.
r/ ) group_by_)Z_groupbyr )r: r r r r1 r1 r2 group_by s $ r )r: r duration_mapperr r. c C s ddl m} || |||S )a Groups the elements of an observable sequence according to a
specified key mapper function. A duration mapper function is used
to control the lifetime of groups. When a group expires, it
receives an OnCompleted notification. When a new element with the
same key value as a reclaimed group occurs, the group will be
reborn with a new lifetime request.
.. marble::
:alt: group_by_until
--1--2--a--3--b--c-|
[ group_by_until() ]
-+-----+-----------|
+a-----b--c-|
+1--2-----3-------|
Examples:
>>> group_by_until(lambda x: x.id, None, lambda : reactivex.never())
>>> group_by_until(
lambda x: x.id, lambda x: x.name, lambda grp: reactivex.never()
)
>>> group_by_until(
lambda x: x.id,
lambda x: x.name,
lambda grp: reactivex.never(),
lambda: ReplaySubject()
)
Args:
key_mapper: A function to extract the key for each element.
element_mapper: A function to map each source element to an element in
an observable group.
duration_mapper: A function to signal the expiration of a group.
subject_mapper: A function that returns a subject used to initiate
a grouped observable. Default mapper returns a Subject object.
Returns:
An operator function that takes an observable source and
returns a sequence of observable groups, each of which
corresponds to a unique key value, containing all elements that
share that same key value. If a group's lifetime expires, a new
group with the same key value can be created once an element
with such a key value is encountered.
r/ )group_by_until_)Z
_groupbyuntilr )r: r r r r r1 r1 r2 group_by_until s 2 r )rightleft_duration_mapperright_duration_mapperr. c C s ddl m} || ||S )a Correlates the elements of two sequences based on overlapping
durations, and groups the results.
.. marble::
:alt: group_join
-1---2----3---4---->
--a--------b-----c->
[ group_join() ]
--a1-a2----b3-b4-c4|
Args:
right: The right observable sequence to join elements for.
left_duration_mapper: A function to select the duration
(expressed as an observable sequence) of each element of
the left observable sequence, used to determine overlap.
right_duration_mapper: A function to select the duration
(expressed as an observable sequence) of each element of
the right observable sequence, used to determine overlap.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains elements combined into
a tuple from source elements that have an overlapping
duration.
r/ ) group_join_)Z
_groupjoinr )r r r r r1 r1 r2
group_join
s r c C s ddl m} | S )a Ignores all elements in an observable sequence leaving only the
termination messages.
.. marble::
:alt: ignore_elements
---1---2---3---4---|
[ ignore_elements()]
-------------------|
Returns:
An operator function that takes an observable source and
returns an empty observable sequence that signals termination,
successful or exceptional, of the source sequence.
r/ ignore_elements_)Z_ignoreelementsr r r1 r1 r2 ignore_elements/ s r c C s ddl m} | S )ar Determines whether an observable sequence is empty.
.. marble::
:alt: is_empty
-------|
[ is_empty() ]
-------True-|
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element
determining whether the source sequence is empty.
r/ is_empty_)Z_isemptyr r r1 r1 r2 is_emptyD s r c C s ddl m} || ||S )a Correlates the elements of two sequences based on overlapping
durations.
.. marble::
:alt: join
-1---2----3---4---->
--a--------b-----c->
[ join() ]
--a1-a2----b3-b4-c4|
Args:
right: The right observable sequence to join elements for.
left_duration_mapper: A function to select the duration
(expressed as an observable sequence) of each element of
the left observable sequence, used to determine overlap.
right_duration_mapper: A function to select the duration
(expressed as an observable sequence) of each element of
the right observable sequence, used to determine overlap.
Return:
An operator function that takes an observable source and
returns an observable sequence that contains elements combined
into a tuple from source elements that have an overlapping
duration.
r/ )join_)Z_joinr )r r r r r1 r1 r2 joinY s r c C s ddl m} || S )a The last operator.
Returns the last element of an observable sequence that satisfies
the condition in the predicate if specified, else the last element.
.. marble::
:alt: last
---1--2--3--4-|
[ last() ]
------------4-|
Examples:
>>> op = last()
>>> op = last(lambda x: x > 3)
Args:
predicate: [Optional] A predicate function to evaluate for
elements in the source sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the last element in
the observable sequence that satisfies the condition in the
predicate.
r/ )last_)_lastr )r- r r1 r1 r2 last} s r c C s d S rm r1 r1 r1 r1 r2 last_or_default s r c C s d S rm r1 rn r1 r1 r2 r s )rl r- r. c C s d S rm r1 )rl r- r1 r1 r2 r s c C s ddl m} || |S )a The last_or_default operator.
Returns the last element of an observable sequence that satisfies
the condition in the predicate, or a default value if no such
element exists.
.. marble::
:alt: last
---1--2--3--4-|
[last_or_default(8)]
--------------8-|
Examples:
>>> res = last_or_default()
>>> res = last_or_default(lambda x: x > 3)
>>> res = last_or_default(lambda x: x > 3, 0)
>>> res = last_or_default(None, 0)
Args:
predicate: [Optional] A predicate function to evaluate for
elements in the source sequence.
default_value: [Optional] The default value if no such element
exists. If not specified, defaults to None.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the last element in
the observable sequence that satisfies the condition in the
predicate, or a default value if no such element exists.
r/ )r )Z_lastordefaultr )rl r- r r1 r1 r2 r s $ c C s ddl m} || S )aW The map operator.
Project each element of an observable sequence into a new form.
.. marble::
:alt: map
---1---2---3---4--->
[ map(i: i*2) ]
---2---4---6---8--->
Example:
>>> map(lambda value: value * 10)
Args:
mapper: A transform function to apply to each source element.
Returns:
A partially applied operator function that takes an observable
source and returns an observable sequence whose elements are
the result of invoking the transform function on each element
of the source.
r/ map_)_mapr )r r r1 r1 r2 r^ s r^ c C s ddl m} || S )a Project each element of an observable sequence into a new form
by incorporating the element's index.
.. marble::
:alt: map_indexed
---1---2---3---4--->
[ map(i,id: i*2) ]
---2---4---6---8--->
Example:
>>> ret = map_indexed(lambda value, index: value * value + index)
Args:
mapper_indexed: A transform function to apply to each source
element. The second parameter of the function represents
the index of the source element.
Returns:
A partially applied operator function that takes an observable
source and returns an observable sequence whose elements are
the result of invoking the transform function on each element
of the source.
r/ ) map_indexed_)r r )r r r1 r1 r2 map_indexed s r c C s ddl m} | S )a3 Materializes the implicit notifications of an observable
sequence as explicit notification values.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the materialized
notification values from the source sequence.
r/ materialize)Z _materializer r r1 r1 r2 r s r )rc r. c C s ddl m} || S )aW Returns the maximum value in an observable sequence according to
the specified comparer.
.. marble::
:alt: max
---1--2--3--4-|
[ max() ]
--------------4-|
Examples:
>>> op = max()
>>> op = max(lambda x, y: x.value - y.value)
Args:
comparer: [Optional] Comparer used to compare elements.
Returns:
A partially applied operator function that takes an observable
source and returns an observable sequence containing a single
element with the maximum element in the source sequence.
r/ )max_)_maxr )rc r r1 r1 r2 max' s r c C s ddl m} || |S )a The max_by operator.
Returns the elements in an observable sequence with the maximum
key value according to the specified comparer.
.. marble::
:alt: max_by
---1--2--3--4-|
[ max_by() ]
--------------4-|
Examples:
>>> res = max_by(lambda x: x.value)
>>> res = max_by(lambda x: x.value, lambda x, y: x - y)
Args:
key_mapper: Key mapper function.
comparer: [Optional] Comparer used to compare key values.
Returns:
A partially applied operator function that takes an observable
source and return an observable sequence containing a list of
zero or more elements that have a maximum key value.
r/ )max_by_)Z_maxbyr )r: rc r r1 r1 r2 max_byE s r r\ )rX r] r. c G s ddl m} ||d| iS )aD Merges an observable sequence of observable sequences into an
observable sequence, limiting the number of concurrent
subscriptions to inner sequences. Or merges two observable
sequences into a single observable sequence.
.. marble::
:alt: merge
---1---2---3---4-|
-a---b---c---d--|
[ merge() ]
-a-1-b-2-c-3-d-4-|
Examples:
>>> op = merge(max_concurrent=1)
>>> op = merge(other_source)
Args:
max_concurrent: [Optional] Maximum number of inner observable
sequences being subscribed to concurrently or the second
observable sequence.
Returns:
An operator function that takes an observable source and
returns the observable sequence that merges the elements of the
inner sequences.
r/ )merge_r] )_merger )r] rX r r1 r1 r2 r_ f s r_ c C s ddl m} | S )a The merge_all operator.
Merges an observable sequence of observable sequences into an
observable sequence.
.. marble::
:alt: merge_all
---1---2---3---4-|
-a---b---c---d--|
[ merge_all() ]
-a-1-b-2-c-3-d-4-|
Returns:
A partially applied operator function that takes an observable
source and returns the observable sequence that merges the
elements of the inner sequences.
r/
merge_all_)r r r r1 r1 r2 merge_all s r c C s ddl m} || S )a The `min` operator.
Returns the minimum element in an observable sequence according to
the optional comparer else a default greater than less than check.
.. marble::
:alt: min
---1--2--3--4-|
[ min() ]
--------------1-|
Examples:
>>> res = source.min()
>>> res = source.min(lambda x, y: x.value - y.value)
Args:
comparer: [Optional] Comparer used to compare elements.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element
with the minimum element in the source sequence.
r/ )min_)_minr )rc r r1 r1 r2 min s r c C s ddl m} || |S )a The `min_by` operator.
Returns the elements in an observable sequence with the minimum key
value according to the specified comparer.
.. marble::
:alt: min_by
---1--2--3--4-|
[ min_by() ]
--------------1-|
Examples:
>>> res = min_by(lambda x: x.value)
>>> res = min_by(lambda x: x.value, lambda x, y: x - y)
Args:
key_mapper: Key mapper function.
comparer: [Optional] Comparer used to compare key values.
Returns:
An operator function that takes an observable source and
reuturns an observable sequence containing a list of zero
or more elements that have a minimum key value.
r/ )min_by_)Z_minbyr )r: rc r r1 r1 r2 min_by s r c C s d S rm r1 r1 r1 r1 r2 multicast s r )subjectr. c C s d S rm r1 )r r1 r1 r2 r s r )subject_factoryr r. c C s d S rm r1 r r r1 r1 r2 r s r )r r r r. c C s ddl m} || ||dS )a Multicasts the source sequence notifications through an
instantiated subject into all uses of the sequence within a mapper
function. Each subscription to the resulting sequence causes a
separate multicast invocation, exposing the sequence resulting from
the mapper function's invocation. For specializations with fixed
subject types, see Publish, PublishLast, and Replay.
Examples:
>>> res = multicast(observable)
>>> res = multicast(
subject_factory=lambda scheduler: Subject(), mapper=lambda x: x
)
Args:
subject_factory: Factory function to create an intermediate
subject through which the source sequence's elements will
be multicast to the mapper function.
subject: Subject to push source elements into.
mapper: [Optional] Mapper function which can use the
multicasted source sequence subject to the policies
enforced by the created subject. Specified only if
subject_factory" is a factory function.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements of a
sequence produced by multicasting the source sequence within a
mapper function.
r/ )
multicast_r )Z
_multicastr )r r r r r1 r1 r2 r s % )rM r. c C s ddl m} || S )a Wraps the source sequence in order to run its observer callbacks
on the specified scheduler.
Args:
scheduler: Scheduler to notify observers on.
This only invokes observer callbacks on a scheduler. In case the
subscription and/or unsubscription actions have side-effects
that require to be run on a scheduler, use subscribe_on.
Returns:
An operator function that takes an observable source and
returns the source sequence whose observations happen on the
specified scheduler.
r/ ) observe_on_)Z
_observeonr )rM r r1 r1 r2
observe_on s r )secondr. c C s ddl m} || S )a: Continues an observable sequence that is terminated normally
or by an exception with the next observable sequence.
.. marble::
:alt: on_error
---1--2--3--4-*
e-a--b-|
[ on_error(e) ]
-1--2--3--4-a--b-|
Keyword arguments:
second: Second observable sequence used to produce results
after the first sequence terminates.
Returns:
An observable sequence that concatenates the first and
second sequence, even if the first sequence terminates
exceptionally.
r/ )on_error_resume_next_)Z_onerrorresumenextr )r r r1 r1 r2 on_error_resume_next5 s r c C s ddl m} | S )aB The pairwise operator.
Returns a new observable that triggers on the second and subsequent
triggerings of the input observable. The Nth triggering of the
input observable passes the arguments from the N-1th and Nth
triggering as a pair. The argument passed to the N-1th triggering
is held in hidden internal state until the Nth triggering occurs.
Returns:
An operator function that takes an observable source and
returns an observable that triggers on successive pairs of
observations from the input observable as an array.
r/ pairwise_)Z _pairwiser r r1 r1 r2 pairwiseS s r c C s ddl m} || S )a Returns two observables which partition the observations of the
source by the given function. The first will trigger observations
for those values for which the predicate returns true. The second
will trigger observations for those values where the predicate
returns false. The predicate is executed once for each subscribed
observer. Both also propagate all error observations arising from
the source and each completes when the source completes.
.. marble::
:alt: partition
---1--2--3--4--|
[ partition(even) ]
---1-----3-----|
------2-----4--|
Args:
predicate: The function to determine which output Observable
will trigger a particular observation.
Returns:
An operator function that takes an observable source and
returns a list of observables. The first triggers when the
predicate returns True, and the second triggers when the
predicate returns False.
r/ )
partition_)
_partitionr )r- r r1 r1 r2 partitionf s r c C s ddl m} || S )a The indexed partition operator.
Returns two observables which partition the observations of the
source by the given function. The first will trigger observations
for those values for which the predicate returns true. The second
will trigger observations for those values where the predicate
returns false. The predicate is executed once for each subscribed
observer. Both also propagate all error observations arising from
the source and each completes when the source completes.
.. marble::
:alt: partition_indexed
---1--2--3--4--|
[ partition(even) ]
---1-----3-----|
------2-----4--|
Args:
predicate: The function to determine which output Observable
will trigger a particular observation.
Returns:
A list of observables. The first triggers when the predicate
returns True, and the second triggers when the predicate
returns False.
r/ )partition_indexed_)r r )r r r1 r1 r2 partition_indexed s r )keyr. c C s ddl m} || S )aw Retrieves the value of a specified key using dict-like access (as in
element[key]) from all elements in the Observable sequence.
To pluck an attribute of each element, use pluck_attr.
Args:
key: The key to pluck.
Returns:
An operator function that takes an observable source and
returns a new observable sequence of key values.
r/ )pluck_)_pluckr )r r r1 r1 r2 pluck s r )propr. c C s ddl m} || S )a Retrieves the value of a specified property (using getattr) from
all elements in the Observable sequence.
To pluck values using dict-like access (as in element[key]) on each
element, use pluck.
Args:
property: The property to pluck.
Returns:
An operator function that takes an observable source and
returns a new observable sequence of property values.
r/ ) pluck_attr_)r r )r r r1 r1 r2
pluck_attr s r c C s d S rm r1 r1 r1 r1 r2 publish s r c C s d S rm r1 r r1 r1 r2 r s c C s ddl m} || S )a The `publish` operator.
Returns an observable sequence that is the result of invoking the
mapper on a connectable observable sequence that shares a single
subscription to the underlying sequence. This operator is a
specialization of Multicast using a regular Subject.
Example:
>>> res = publish()
>>> res = publish(lambda x: x)
Args:
mapper: [Optional] Selector function which can use the
multicasted source sequence as many times as needed,
without causing multiple subscriptions to the source
sequence. Subscribers to the given source will receive all
notifications of the source from the time of the
subscription on.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements of a
sequence produced by multicasting the source sequence within a
mapper function.
r/ )publish_)_publishr )r r r1 r1 r2 r s )
initial_valuer. c C s d S rm r1 )r r1 r1 r2
publish_value s r )r r r. c C s d S rm r1 )r r r1 r1 r2 r s c C s ddl m} || |S )a Returns an observable sequence that is the result of invoking
the mapper on a connectable observable sequence that shares a
single subscription to the underlying sequence and starts with
initial_value.
This operator is a specialization of Multicast using a
BehaviorSubject.
Examples:
>>> res = source.publish_value(42)
>>> res = source.publish_value(42, lambda x: x.map(lambda y: y * y))
Args:
initial_value: Initial value received by observers upon
subscription.
mapper: [Optional] Optional mapper function which can use the
multicasted source sequence as many times as needed,
without causing multiple subscriptions to the source
sequence. Subscribers to the given source will receive
immediately receive the initial value, followed by all
notifications of the source from the time of the
subscription on.
Returns:
An operator function that takes an observable source and returns
an observable sequence that contains the elements of a
sequence produced by multicasting the source sequence within a
mapper function.
r/ )publish_value_)Z
_publishvaluer )r r r r1 r1 r2 r
s $ ) accumulatorr. c C s d S rm r1 r r1 r1 r2 reduce3 s r )r seedr. c C s d S rm r1 r r
r1 r1 r2 r 9 s c C s ddl m} || |S )a The reduce operator.
Applies an accumulator function over an observable sequence,
returning the result of the aggregation as a single element in the
result sequence. The specified seed value is used as the initial
accumulator value.
For aggregation behavior with incremental intermediate results,
see `scan`.
.. marble::
:alt: reduce
---1--2--3--4--|
[reduce(acc,i: acc+i)]
---------------10-|
Examples:
>>> res = reduce(lambda acc, x: acc + x)
>>> res = reduce(lambda acc, x: acc + x, 0)
Args:
accumulator: An accumulator function to be invoked on each
element.
seed: Optional initial accumulator value.
Returns:
A partially applied operator function that takes an observable
source and returns an observable sequence containing a single
element with the final accumulator value.
r/ )reduce_)Z_reducer )r r
r r1 r1 r2 r ? s " c C s ddl m} | S )zReturns an observable sequence that stays connected to the
source as long as there is at least one subscription to the
observable sequence.
r/
ref_count_)Zconnectable._refcountr r
r1 r1 r2 ref_countf s r ) repeat_countr. c C s ddl m} || S )a Repeats the observable sequence a specified number of times.
If the repeat count is not specified, the sequence repeats
indefinitely.
.. marble::
:alt: repeat
-1--2-|
[ repeat(3) ]
-1--2--1--2--1--2-|
Examples:
>>> repeated = repeat()
>>> repeated = repeat(42)
Args:
repeat_count: Number of times to repeat the sequence. If not
provided, repeats the sequence indefinitely.
Returns:
An operator function that takes an observable sources and
returns an observable sequence producing the elements of the
given sequence repeatedly.
r/ )repeat_)_repeatr )r r r1 r1 r2 repeatp s r rr ) buffer_sizewindowrM r. c C s d S rm r1 )r r rM r1 r1 r2 replay s r )r r r rM r. c C s d S rm r1 )r r r rM r1 r1 r2 r s )r rM c C s ddl m} ||| ||dS )a The `replay` operator.
Returns an observable sequence that is the result of invoking the
mapper on a connectable observable sequence that shares a single
subscription to the underlying sequence replaying notifications
subject to a maximum time length for the replay buffer.
This operator is a specialization of Multicast using a
ReplaySubject.
Examples:
>>> res = replay(buffer_size=3)
>>> res = replay(buffer_size=3, window=0.5)
>>> res = replay(None, 3, 0.5)
>>> res = replay(lambda x: x.take(6).repeat(), 3, 0.5)
Args:
mapper: [Optional] Selector function which can use the
multicasted source sequence as many times as needed,
without causing multiple subscriptions to the source
sequence. Subscribers to the given source will receive all
the notifications of the source subject to the specified
replay buffer trimming policy.
buffer_size: [Optional] Maximum element count of the replay
buffer.
window: [Optional] Maximum time length of the replay buffer.
scheduler: [Optional] Scheduler the observers are invoked on.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements of a
sequence produced by multicasting the source sequence within a
mapper function.
r/ )replay_rr )Z_replayr )r r r rM r r1 r1 r2 r s ) ) retry_countr. c C s ddl m} || S )a Repeats the source observable sequence the specified number of
times or until it successfully terminates. If the retry count is
not specified, it retries indefinitely.
Examples:
>>> retried = retry()
>>> retried = retry(42)
Args:
retry_count: [Optional] Number of times to retry the sequence.
If not provided, retry the sequence indefinitely.
Returns:
An observable sequence producing the elements of the given
sequence repeatedly until it terminates successfully.
r/ )retry_)Z_retryr )r r r1 r1 r2 retry s r )samplerrM r. c C s ddl m} || |S )a Samples the observable sequence at each interval.
.. marble::
:alt: sample
---1-2-3-4------|
[ sample(4) ]
----1---3---4---|
Examples:
>>> res = sample(sample_observable) # Sampler tick sequence
>>> res = sample(5.0) # 5 seconds
Args:
sampler: Observable used to sample the source observable **or** time
interval at which to sample (specified as a float denoting
seconds or an instance of timedelta).
scheduler: Scheduler to use only when a time interval is given.
Returns:
An operator function that takes an observable source and
returns a sampled observable sequence.
r/ )sample_)Z_sampler )r rM r r1 r1 r2 sample s r c C s d S rm r1 r r1 r1 r2 scan
s r c C s d S rm r1 r r1 r1 r2 r
s c C s ddl m} || |S )aw The scan operator.
Applies an accumulator function over an observable sequence and
returns each intermediate result. The optional seed value is used
as the initial accumulator value. For aggregation behavior with no
intermediate results, see `aggregate()` or `Observable()`.
.. marble::
:alt: scan
----1--2--3--4-----|
[scan(acc,i: acc+i)]
----1--3--6--10----|
Examples:
>>> scanned = source.scan(lambda acc, x: acc + x)
>>> scanned = source.scan(lambda acc, x: acc + x, 0)
Args:
accumulator: An accumulator function to be invoked on each
element.
seed: [Optional] The initial accumulator value.
Returns:
A partially applied operator function that takes an observable
source and returns an observable sequence containing the
accumulated values.
r/ )scan_)Z_scanr )r r
r r1 r1 r2 r
s )r rc r. c C s ddl m} || |S )aj Determines whether two sequences are equal by comparing the
elements pairwise using a specified equality comparer.
.. marble::
:alt: scan
-1--2--3--4----|
----1--2--3--4-|
[ sequence_equal() ]
---------------True|
Examples:
>>> res = sequence_equal([1,2,3])
>>> res = sequence_equal([{ "value": 42 }], lambda x, y: x.value == y.value)
>>> res = sequence_equal(reactivex.return_value(42))
>>> res = sequence_equal(
reactivex.return_value({ "value": 42 }), lambda x, y: x.value == y.value)
Args:
second: Second observable sequence or iterable to compare.
comparer: [Optional] Comparer used to compare elements of both
sequences. No guarantees on order of comparer arguments.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains a single element
which indicates whether both sequences are of equal length and
their corresponding elements are equal according to the
specified equality comparer.
r/ )sequence_equal_)Z_sequenceequalr )r rc r r1 r1 r2 sequence_equal9
s ! r! c C s ddl m} | S )a Share a single subscription among multiple observers.
This is an alias for a composed publish() and ref_count().
Returns:
An operator function that takes an observable source and
returns a new Observable that multicasts (shares) the original
Observable. As long as there is at least one Subscriber this
Observable will be subscribed and emitting data. When all
subscribers have unsubscribed it will unsubscribe from the
source
Observable.
r/ share_)r r# r" r1 r1 r2 share_
s r$ c C s ddl m} || S )a& The single operator.
Returns the only element of an observable sequence that satisfies
the condition in the optional predicate, and reports an exception
if there is not exactly one element in the observable sequence.
.. marble::
:alt: single
----1--2--3--4-----|
[ single(3) ]
----------3--------|
Example:
>>> res = single()
>>> res = single(lambda x: x == 42)
Args:
predicate: [Optional] A predicate function to evaluate for
elements in the source sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the single element in
the observable sequence that satisfies the condition in the
predicate.
r/ )single_)Z_singler% )r- r% r1 r1 r2 singler
s r& c C s ddl m} || |S )aV Returns the only element of an observable sequence that matches
the predicate, or a default value if no such element exists this
method reports an exception if there is more than one element in
the observable sequence.
.. marble::
:alt: single_or_default
----1--2--3--4--|
[ single(8,42) ]
----------------42-|
Examples:
>>> res = single_or_default()
>>> res = single_or_default(lambda x: x == 42)
>>> res = single_or_default(lambda x: x == 42, 0)
>>> res = single_or_default(None, 0)
Args:
predicate: [Optional] A predicate function to evaluate for
elements in the source sequence.
default_value: [Optional] The default value if the index is
outside the bounds of the source sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the single element in
the observable sequence that satisfies the condition in the
predicate, or a default value if no such element exists.
r/ )single_or_default_)_singleordefaultr' )r- rl r' r1 r1 r2 single_or_default
s ! r) F) has_defaultrl r. c C s ddl m} || |S )Nr/ )single_or_default_async_)r( r+ )r* rl r+ r1 r1 r2 single_or_default_async
s r, )rG r. c C s ddl m} || S )aK The skip operator.
Bypasses a specified number of elements in an observable sequence
and then returns the remaining elements.
.. marble::
:alt: skip
----1--2--3--4-----|
[ skip(2) ]
----------3--4-----|
Args:
count: The number of elements to skip before returning the
remaining elements.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements that
occur after the specified index in the input sequence.
r/ )skip_)Z_skipr- )rG r- r1 r1 r2 rH
s rH c C s ddl m} || S )a* The skip_last operator.
.. marble::
:alt: skip_last
----1--2--3--4-----|
[ skip_last(1) ]
-------1--2--3-----|
Bypasses a specified number of elements at the end of an observable
sequence.
This operator accumulates a queue with a length enough to store the
first `count` elements. As more elements are received, elements are
taken from the front of the queue and produced on the result
sequence. This causes elements to be delayed.
Args:
count: Number of elements to bypass at the end of the source
sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the source sequence
elements except for the bypassed ones at the end.
r/ )
skip_last_)Z _skiplastr. )rG r. r1 r1 r2 skip_last
s r/ )durationrM r. c C s ddl m} || |dS )a Skips elements for the specified duration from the end of the
observable source sequence.
Example:
>>> res = skip_last_with_time(5.0)
This operator accumulates a queue with a length enough to store
elements received during the initial duration window. As more
elements are received, elements older than the specified duration
are taken from the queue and produced on the result sequence. This
causes elements to be delayed with duration.
Args:
duration: Duration for skipping elements from the end of the
sequence.
scheduler: Scheduler to use for time handling.
Returns:
An observable sequence with the elements skipped during the
specified duration from the end of the source sequence.
r/ )skip_last_with_time_rr )Z_skiplastwithtimer1 )r0 rM r1 r1 r1 r2 skip_last_with_time s r2 z Future[Any])otherr. c C s ddl m} || S )a Returns the values from the source observable sequence only
after the other observable sequence produces a value.
.. marble::
:alt: skip_until
----1--2--3--4-----|
---------1---------|
[ skip_until() ]
----------3--4-----|
Args:
other: The observable sequence that triggers propagation of
elements of the source sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the elements of the
source sequence starting from the point the other sequence
triggered propagation.
r/ ) skip_until_)Z
_skipuntilr4 )r3 r4 r1 r1 r2
skip_until s r5 )
start_timerM r. c C s ddl m} || |dS )al Skips elements from the observable source sequence until the
specified start time.
Errors produced by the source sequence are always forwarded to the
result sequence, even if the error occurs before the start time.
.. marble::
:alt: skip_until
------1--2--3--4-------|
[skip_until_with_time()]
------------3--4-------|
Examples:
>>> res = skip_until_with_time(datetime())
>>> res = skip_until_with_time(5.0)
Args:
start_time: Time to start taking elements from the source
sequence. If this value is less than or equal to
`datetime.now(timezone.utc)`, no elements will be skipped.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the elements skipped
until the specified start time.
r/ )skip_until_with_time_rr )Z_skipuntilwithtimer7 )r6 rM r7 r1 r1 r2 skip_until_with_time: s r8 c C s ddl m} || S )a The `skip_while` operator.
Bypasses elements in an observable sequence as long as a specified
condition is true and then returns the remaining elements. The
element's index is used in the logic of the predicate function.
.. marble::
:alt: skip_while
----1--2--3--4-----|
[skip_while(i: i<3)]
----------3--4-----|
Example:
>>> skip_while(lambda value: value < 10)
Args:
predicate: A function to test each element for a condition; the
second parameter of the function represents the index of
the source element.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements from
the input sequence starting at the first element in the linear
series that does not pass the test specified by predicate.
r/ ) skip_while_)
_skipwhiler9 )r- r9 r1 r1 r2
skip_while] s r; c C s ddl m} || S )a Bypasses elements in an observable sequence as long as a
specified condition is true and then returns the remaining
elements. The element's index is used in the logic of the predicate
function.
.. marble::
:alt: skip_while_indexed
----1--2--3--4-----|
[skip_while(i: i<3)]
----------3--4-----|
Example:
>>> skip_while(lambda value, index: value < 10 or index < 10)
Args:
predicate: A function to test each element for a condition; the
second parameter of the function represents the index of
the source element.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements from
the input sequence starting at the first element in the linear
series that does not pass the test specified by predicate.
r/ )skip_while_indexed_)r: r< )r- r< r1 r1 r2 skip_while_indexed s r= c C s ddl m} || |dS )aC Skips elements for the specified duration from the start of the
observable source sequence.
.. marble::
:alt: skip_with_time
----1--2--3--4-----|
[ skip_with_time() ]
----------3--4-----|
Args:
>>> res = skip_with_time(5.0)
Specifying a zero value for duration doesn't guarantee no elements
will be dropped from the start of the source sequence. This is a
side-effect of the asynchrony introduced by the scheduler, where
the action that causes callbacks from the source sequence to be
forwarded may not execute immediately, despite the zero due time.
Errors produced by the source sequence are always forwarded to the
result sequence, even if the error occurs before the duration.
Args:
duration: Duration for skipping elements from the start of the
sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the elements skipped during
the specified duration from the start of the source sequence.
r/ )skip_with_time_rr )Z
_skipwithtimer> )r0 rM r> r1 r1 r2 skip_with_time s " r? )startstopstepr. c C s ddl m} || ||S )a The slice operator.
Slices the given observable. It is basically a wrapper around the operators
:func:`skip `,
:func:`skip_last `,
:func:`take `,
:func:`take_last ` and
:func:`filter `.
.. marble::
:alt: slice
----1--2--3--4-----|
[ slice(1, 2) ]
-------2--3--------|
Examples:
>>> result = source.slice(1, 10)
>>> result = source.slice(1, -2)
>>> result = source.slice(1, -1, 2)
Args:
start: First element to take of skip last
stop: Last element to take of skip last
step: Takes every step element. Must be larger than zero
Returns:
An operator function that takes an observable source and
returns a sliced observable sequence.
r/ )slice_)Z_slicerC )r@ rA rB rC r1 r1 r2 slice s ! rD c C s ddl m} || S )a The some operator.
Determines whether some element of an observable sequence
satisfies a condition if present, else if some items are in the
sequence.
.. marble::
:alt: some
----1--2--3--4-----|
[ some(i: i>3) ]
-------------True--|
Examples:
>>> result = source.some()
>>> result = source.some(lambda x: x > 3)
Args:
predicate: A function to test each element for a condition.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element
determining whether some elements in the source sequence
pass the test in the specified predicate if given, else if some
items are in the sequence.
r/ )some_)Z_somerE )r- rE r1 r1 r2 some s rF c C s d S rm r1 r r1 r1 r2 starmap s rG c C s d S rm r1 r r1 r1 r2 rG s c C s d S rm r1 r r1 r1 r2 rG s c s8 dkrt tS ttdf td fdd }t t|S )a The starmap operator.
Unpack arguments grouped as tuple elements of an observable
sequence and return an observable sequence of values by invoking
the mapper function with star applied unpacked elements as
positional arguments.
Use instead of `map()` when the the arguments to the mapper is
grouped as tuples and the mapper function takes multiple arguments.
.. marble::
:alt: starmap
-----1,2---3,4-----|
[ starmap(add) ]
-----3-----7-------|
Example:
>>> starmap(lambda x, y: x + y)
Args:
mapper: A transform function to invoke with unpacked elements
as arguments.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the results of
invoking the mapper function with unpacked elements of the
source.
N.valuesr. c s st | S rm AssertionErrorrI r r1 r2 starredI s zstarmap..starred)r r r r r^ )r rM r1 r r2 rG $ s " c C s d S rm r1 r r1 r1 r2 starmap_indexedP s rN c C s d S rm r1 r r1 r1 r2 rN V s c C s d S rm r1 r r1 r1 r2 rN \ s c C s d S rm r1 r r1 r1 r2 rN b s c sD ddl m} dkr ttS ttdf td fdd }t||S )az Variant of :func:`starmap` which accepts an indexed mapper.
.. marble::
:alt: starmap_indexed
---------1,2---3,4---------|
[ starmap_indexed(sum) ]
---------3-----8-----------|
Example:
>>> starmap_indexed(lambda x, y, i: x + y + i)
Args:
mapper: A transform function to invoke with unpacked elements
as arguments.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the results of
invoking the indexed mapper function with unpacked elements
of the source.
r/ r N.rH c s st | S rm rJ rL r r1 r2 rM s z starmap_indexed..starred)r r r r r r )r r rM r1 r r2 rN h s
)argsr. c G s ddl m} || S )a Prepends a sequence of values to an observable sequence.
.. marble::
:alt: start_with
-----1--2--3--4----|
[ start_with(7,8) ]
-7-8-1--2--3--4----|
Example:
>>> start_with(1, 2, 3)
Returns:
An operator function that takes a source observable and returns
the source sequence prepended with the specified values.
r/ ) start_with_)Z _startswithrP )rO rP r1 r1 r2
start_with s rQ c C s ddl m} || S )a Subscribe on the specified scheduler.
Wrap the source sequence in order to run its subscription and
unsubscription logic on the specified scheduler. This operation is
not commonly used; see the remarks section for more information on
the distinction between subscribe_on and observe_on.
This only performs the side-effects of subscription and
unsubscription on the specified scheduler. In order to invoke
observer callbacks on a scheduler, use observe_on.
Args:
scheduler: Scheduler to perform subscription and unsubscription
actions on.
Returns:
An operator function that takes an observable source and
returns the source sequence whose subscriptions and
un-subscriptions happen on the specified scheduler.
r/ )
subscribe_on_)Z _subscribeonrR )rM rR r1 r1 r2 subscribe_on s rS c C s d S rm r1 r1 r1 r1 r2 sum s rT c C s d S rm r1 )r: r1 r1 r2 rT s c C s ddl m} || S )a Computes the sum of a sequence of values that are obtained by
invoking an optional transform function on each element of the
input sequence, else if not specified computes the sum on each item
in the sequence.
.. marble::
:alt: sum
-----1--2--3--4-|
[ sum() ]
----------------10-|
Examples:
>>> res = sum()
>>> res = sum(lambda x: x.value)
Args:
key_mapper: [Optional] A transform function to apply to each
element.
Returns:
An operator function that takes a source observable and returns
an observable sequence containing a single element with the sum
of the values in the source sequence.
r/ )sum_)Z_sumrU )r: rU r1 r1 r2 rT s z
Future[_T]c C s ddl m} | S )a The switch_latest operator.
Transforms an observable sequence of observable sequences into an
observable sequence producing values only from the most recent
observable sequence.
.. marble::
:alt: switch_latest
-+------+----------|
+--a--b--c-|
+--1--2--3--4--|
[ switch_latest() ]
----1--2---a--b--c-|
Returns:
A partially applied operator function that takes an observable
source and returns the observable sequence that at any point in
time produces the elements of the most recent inner observable
sequence that has been received.
r/ switch_latest_)Z
_switchlatestrW rV r1 r1 r2
switch_latest s rX c C s t t| t S )aU Projects each source value to an Observable which is merged in
the output Observable, emitting values only from the most recently
projected Observable.
.. marble::
:alt: switch_map
---a----------b-------c---------|
[ switch_map(x: x---x---x|) ]
---a---a---a--b---b---c---c---c-|
Examples:
>>> op = switch_map(lambda x: reactivex.timer(1.0).pipe(map(lambda x: x)))
>>> op = switch_map()
Args:
project: Projecting function which takes the outer observable value
and the emission index and emits the inner observable; defaults to `identity`
Returns:
An operator function that maps each value to the inner observable
and emits its values in order, emitting values only from the
most recently projected Observable.
If an inner observable complete, the resulting sequence does *not*
complete.
If an inner observable errors, the resulting sequence errors as well.
If the outer observable completes/errors, the resulting sequence
completes/errors.
)r r^ rX r` r1 r1 r2
switch_map
s %rY c C s t t| t S )aF Projects each source value to an Observable which is merged in
the output Observable, emitting values only from the most recently
projected Observable.
.. marble::
:alt: switch_map
---a----------b-------c---------------------|
[ switch_map_indexed(x,i: x*i---x*i---x*i|) ]
---a---a---a--bb---bb-ccc---ccc---ccc-------|
Examples:
>>> op = switch_map_indexed(lambda x, i: reactivex.timer(1.0).pipe(map(x*i)))
Args:
project: Projecting function which takes the outer observable value
and the emission index and emits the inner observable
Returns:
An operator function that maps each value to the inner observable
and emits its values in order, emitting values only from the
most recently projected Observable.
If an inner observable complete, the resulting sequence does *not*
complete.
If an inner observable errors, the resulting sequence errors as well.
If the outer observable completes/errors, the resulting sequence
completes/errors.
)r r rX r` r1 r1 r2 switch_map_indexed/
s $rZ c C s ddl m} || S )a
Returns a specified number of contiguous elements from the start
of an observable sequence.
.. marble::
:alt: take
-----1--2--3--4----|
[ take(2) ]
-----1--2-|
Example:
>>> op = take(5)
Args:
count: The number of elements to return.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the specified
number of elements from the start of the input sequence.
r/ )take_)Z_taker[ )rG r[ r1 r1 r2 takeV
s r\ c C s ddl m} || S )a" Returns a specified number of contiguous elements from the end
of an observable sequence.
.. marble::
:alt: take_last
-1--2--3--4-|
[ take_last(2) ]
------------3--4-|
Example:
>>> res = take_last(5)
This operator accumulates a buffer with a length enough to store
elements count elements. Upon completion of the source sequence,
this buffer is drained on the result sequence. This causes the
elements to be delayed.
Args:
count: Number of elements to take from the end of the source
sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the specified number
of elements from the end of the source sequence.
r/ )
take_last_)Z _takelastr] )rG r] r1 r1 r2 take_lastq
s r^ c C s ddl m} || S )a The `take_last_buffer` operator.
Returns an array with the specified number of contiguous elements
from the end of an observable sequence.
.. marble::
:alt: take_last_buffer
-----1--2--3--4-|
[take_last_buffer(2)]
----------------3,4-|
Example:
>>> res = source.take_last(5)
This operator accumulates a buffer with a length enough to store
elements count elements. Upon completion of the source sequence,
this buffer is drained on the result sequence. This causes the
elements to be delayed.
Args:
count: Number of elements to take from the end of the source
sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single list with
the specified number of elements from the end of the source
sequence.
r/ )take_last_buffer_)Z_takelastbufferr_ )rG r_ r1 r1 r2 take_last_buffer
s r` c C s ddl m} || |dS )a Returns elements within the specified duration from the end of
the observable source sequence.
.. marble::
:alt: take_last_with_time
-----1--2--3--4-|
[take_last_with_time(3)]
----------------4-|
Example:
>>> res = take_last_with_time(5.0)
This operator accumulates a queue with a length enough to store
elements received during the initial duration window. As more
elements are received, elements older than the specified duration
are taken from the queue and produced on the result sequence. This
causes elements to be delayed with duration.
Args:
duration: Duration for taking elements from the end of the
sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the elements taken
during the specified duration from the end of the source
sequence.
r/ )take_last_with_time_rr )Z_takelastwithtimera )r0 rM ra r1 r1 r2 take_last_with_time
s rb c C s ddl m} || S )a Returns the values from the source observable sequence until the
other observable sequence produces a value.
.. marble::
:alt: take_until
-----1--2--3--4----|
-------------a-|
[ take_until(2) ]
-----1--2--3-------|
Args:
other: Observable sequence that terminates propagation of
elements of the source sequence.
Returns:
An operator function that takes an observable source and
returns as observable sequence containing the elements of the
source sequence up to the point the other sequence interrupted
further propagation.
r/ ) take_until_)Z
_takeuntilrc )r3 rc r1 r1 r2
take_until
s rd )end_timerM r. c C s ddl m} || |dS )aw Takes elements for the specified duration until the specified
end time, using the specified scheduler to run timers.
.. marble::
:alt: take_until_with_time
-----1--2--3--4--------|
[take_until_with_time()]
-----1--2--3-----------|
Examples:
>>> res = take_until_with_time(dt, [optional scheduler])
>>> res = take_until_with_time(5.0, [optional scheduler])
Args:
end_time: Time to stop taking elements from the source
sequence. If this value is less than or equal to
`datetime.now(timezone.utc)`, the result stream will complete
immediately.
scheduler: Scheduler to run the timer on.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the elements taken until
the specified end time.
r/ )take_until_with_time_rr )Z_takeuntilwithtimerf )re rM rf r1 r1 r2 take_until_with_time
s rg )r- inclusiver. c C s ddl m} || |S )a* Returns elements from an observable sequence as long as a
specified condition is true.
.. marble::
:alt: take_while
-----1--2--3--4----|
[take_while(i: i<3)]
-----1--2----------|
Example:
>>> take_while(lambda value: value < 10)
Args:
predicate: A function to test each element for a condition.
inclusive: [Optional] When set to True the value that caused
the predicate function to return False will also be emitted.
If not specified, defaults to False.
Returns:
An operator function that takes an observable source and
returns an observable sequence that contains the elements from
the input sequence that occur before the element at which the
test no longer passes.
r/ ) take_while_)
_takewhileri )r- rh ri r1 r1 r2
take_while s rk c C s ddl m} || |S )a Returns elements from an observable sequence as long as a
specified condition is true. The element's index is used in the
logic of the predicate function.
.. marble::
:alt: take_while_indexed
--------1------2------3------4-------|
[take_while_indexed(v, i: v<4 or i<3)]
--------1------2---------------------|
Example:
>>> take_while_indexed(lambda value, index: value < 10 or index < 10)
Args:
predicate: A function to test each element for a condition; the
second parameter of the function represents the index of the
source element.
inclusive: [Optional] When set to True the value that caused
the predicate function to return False will also be emitted.
If not specified, defaults to False.
Returns:
An observable sequence that contains the elements from the
input sequence that occur before the element at which the test no
longer passes.
r/ )take_while_indexed_)rj rl )r- rh rl r1 r1 r2 take_while_indexed: s rm c C s ddl m} || |dS )a Takes elements for the specified duration from the start of the
observable source sequence.
.. marble::
:alt: take_with_time
-----1--2--3--4----|
[ take_with_time() ]
-----1--2----------|
Example:
>>> res = take_with_time(5.0)
This operator accumulates a queue with a length enough to store
elements received during the initial duration window. As more
elements are received, elements older than the specified duration
are taken from the queue and produced on the result sequence. This
causes elements to be delayed with duration.
Args:
duration: Duration for taking elements from the start of the
sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence with the elements taken during
the specified duration from the start of the source sequence.
r/ )take_with_time_rr )Z
_takewithtimern )r0 rM rn r1 r1 r2 take_with_time] s ro )window_durationrM r. c C s ddl m} || |S )a Returns an Observable that emits only the first item emitted by
the source Observable during sequential time windows of a specified
duration.
Args:
window_duration: time to wait before emitting another item
after emitting the last item.
Returns:
An operator function that takes an observable source and
returns an observable that performs the throttle operation.
r/ )throttle_first_)Z_throttlefirstrq )rp rM rq r1 r1 r2 throttle_first s rr )throttle_duration_mapperr. c C s ddl m} || S )a4 The throttle_with_mapper operator.
Ignores values from an observable sequence which are followed by
another value within a computed throttle duration.
Example:
>>> op = throttle_with_mapper(lambda x: reactivex.timer(x+x))
Args:
throttle_duration_mapper: Mapper function to retrieve an
observable sequence indicating the throttle duration for each
given element.
Returns:
A partially applied operator function that takes an observable
source and returns the throttled observable sequence.
r/ )throttle_with_mapper_)rj rt )rs rt r1 r1 r2 throttle_with_mapper s ru r/ ) Timestampz
Timestamp[_T]c C s ddl m} || dS )a The timestamp operator.
Records the timestamp for each value in an observable sequence.
Examples:
>>> timestamp()
Produces objects with attributes `value` and `timestamp`, where
value is the original value.
Returns:
A partially applied operator function that takes an observable
source and returns an observable sequence with timestamp
information on values.
r/ )
timestamp_rr )
_timestamprw )rM rw r1 r1 r2 timestamp s ry )rh r3 rM r. c C s ddl m} || ||S )a Returns the source observable sequence or the other observable
sequence if duetime elapses.
.. marble::
:alt: timeout
-1--2--------3--4--|
o-6--7-|
[ timeout(3,o) ]
-1--2---6--7----------|
Examples:
>>> res = timeout(5.0)
>>> res = timeout(datetime(), return_value(42))
>>> res = timeout(5.0, return_value(42))
Args:
duetime: Absolute (specified as a datetime object) or relative time
(specified as a float denoting seconds or an instance of timedetla)
when a timeout occurs.
other: Sequence to return in case of a timeout. If not
specified, a timeout error throwing sequence will be used.
scheduler:
Returns:
An operator function that takes and observable source and
returns the source sequence switching to the other sequence in
case of a timeout.
r/ )timeout_)_timeoutrz )rh r3 rM rz r1 r1 r2 timeout s " r| )
first_timeouttimeout_duration_mapperr3 r. c C s ddl m} || ||S )a Returns the source observable sequence, switching to the other
observable sequence if a timeout is signaled.
Examples:
>>> res = timeout_with_mapper(reactivex.timer(0.5))
>>> res = timeout_with_mapper(
reactivex.timer(0.5), lambda x: reactivex.timer(0.2)
)
>>> res = timeout_with_mapper(
reactivex.timer(0.5),
lambda x: reactivex.timer(0.2),
reactivex.return_value(42)
)
Args:
first_timeout: [Optional] Observable sequence that represents
the timeout for the first element. If not provided, this
defaults to reactivex.never().
timeout_duration_mapper: [Optional] Selector to retrieve an
observable sequence that represents the timeout between the
current element and the next element.
other: [Optional] Sequence to return in case of a timeout. If
not provided, this is set to reactivex.throw().
Returns:
An operator function that takes an observable source and
returns the source sequence switching to the other sequence in
case of a timeout.
r/ )timeout_with_mapper_)Z_timeoutwithmapperr )r} r~ r3 r r1 r1 r2 timeout_with_mapper s " r ) TimeIntervalzTimeInterval[_T]c C s ddl m} || dS )a Records the time interval between consecutive values in an
observable sequence.
.. marble::
:alt: time_interval
--1--2-----3---4--|
[ time_interval() ]
-----2-----5---5---|
Examples:
>>> res = time_interval()
Return:
An operator function that takes an observable source and
returns an observable sequence with time interval information
on values.
r/ )time_interval_rr )
_timeintervalr )rM r r1 r1 r2
time_interval s r )r: r r. c C s ddl m} || |S )a6 Converts the observable sequence to a Map if it exists.
Args:
key_mapper: A function which produces the key for the
dictionary.
element_mapper: [Optional] An optional function which produces
the element for the dictionary. If not present, defaults to
the value from the observable sequence.
Returns:
An operator function that takes an observable source and
returns an observable sequence with a single value of a
dictionary containing the values from the observable sequence.
r/ )to_dict_)Z_todictr )r: r r r1 r1 r2 to_dict5 s r ) future_ctorr. c C s ddl m} || S )aT Converts an existing observable sequence to a Future.
Example:
op = to_future(asyncio.Future);
Args:
future_ctor: [Optional] The constructor of the future.
Returns:
An operator function that takes an observable source and returns
a future with the last value from the observable sequence.
r/ )
to_future_)Z _tofuturer )r r r1 r1 r2 to_futureK s r c C s ddl m} | S )aD Creates an iterable from an observable sequence.
There is also an alias called ``to_list``.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing a single element with
an iterable containing all the elements of the source sequence.
r/ to_iterable_)Z _toiterabler r r1 r1 r2 to_iterable_ s
r 皙?)rK rM r. c C s ddl m} ||| dS )aG Convert an observable sequence into a marble diagram string.
Args:
timespan: [Optional] duration of each character in second.
If not specified, defaults to 0.1s.
scheduler: [Optional] The scheduler used to run the the input
sequence on.
Returns:
Observable stream.
r/ )
to_marbles)rM rK )Z
_tomarblesr )rK rM r r1 r1 r2 r q s r c C s ddl m} | S )zConverts the observable sequence to a set.
Returns:
An operator function that takes an observable source and
returns an observable sequence with a single value of a set
containing the values from the observable sequence.
r/ to_set_)Z_tosetr r r1 r1 r2 to_set s r c C s ddl m} || S )ac Repeats source as long as condition holds emulating a while
loop.
Args:
condition: The condition which determines if the source will be
repeated.
Returns:
An operator function that takes an observable source and
returns an observable sequence which is repeated as long as the
condition holds.
r/ ) while_do_)Z_whiledor )r r r1 r1 r2 while_do s r c C s ddl m} || S )a Projects each element of an observable sequence into zero or
more windows.
.. marble::
:alt: window
---a-----b-----c--------|
----1--2--3--4--5--6--7-|
[ window(open) ]
+--+-----+-----+--------|
+5--6--7-|
+3--4-|
+1--2-|
+--|
Examples:
>>> res = window(reactivex.interval(1.0))
Args:
boundaries: Observable sequence whose elements denote the
creation and completion of non-overlapping windows.
Returns:
An operator function that takes an observable source and
returns an observable sequence of windows.
r/ )window_)_windowr )r= r r1 r1 r2 r s r c C s ddl m} || S )aB Projects each element of an observable sequence into zero or
more windows.
.. marble::
:alt: window
------c|
------c|
------c|
----1--2--3--4--5-|
[ window(close) ]
+-----+-----+-----+|
+4--5-|
+2--3-|
+----1|
Examples:
>>> res = window(lambda: reactivex.timer(0.5))
Args:
closing_mapper: A function invoked to define
the closing of each produced window. It defines the
boundaries of the produced windows (a window is started
when the previous one is closed, resulting in
non-overlapping windows).
Returns:
An operator function that takes an observable source and
returns an observable sequence of windows.
r/ ) window_when_)r r )rA r r1 r1 r2 window_when s ! r c C s ddl m} || |S )a Projects each element of an observable sequence into zero or
more windows.
.. marble::
:alt: window
---a-----------b------------|
---d--|
--------e-|
----1--2--3--4--5--6--7--8--|
[ window(open, close) ]
---+-----------+------------|
+5--6--7|
+1-|
>>> res = window(reactivex.interval(0.5), lambda i: reactivex.timer(i))
Args:
openings: Observable sequence whose elements denote the
creation of windows.
closing_mapper: A function invoked to define the closing of each
produced window. Value from openings Observable that initiated
the associated window is provided as argument to the function.
Returns:
An operator function that takes an observable source and
returns an observable sequence of windows.
r/ )window_toggle_)r r )rD rA r r1 r1 r2
window_toggle s r c C s ddl m} || |S )a Projects each element of an observable sequence into zero or more
windows which are produced based on element count information.
.. marble::
:alt: window_with_count
---1-2-3---4-5-6--->
[ window(3) ]
--+-------+-------->
+4-5-6-|
+1-2-3-|
Examples:
>>> window_with_count(10)
>>> window_with_count(10, 1)
Args:
count: Length of each window.
skip: [Optional] Number of elements to skip between creation of
consecutive windows. If not specified, defaults to the
count.
Returns:
An observable sequence of windows.
r/ )window_with_count_)Z_windowwithcountr )rG rH r r1 r1 r2 window_with_count s r c C s ddl m} || ||S )Nr/ )window_with_time_)Z_windowwithtimer )rK rL rM r r1 r1 r2 window_with_time3 s r c C s ddl m} || ||S )Nr/ )window_with_time_or_count_)Z_windowwithtimeorcountr )rK rG rM r r1 r1 r2 window_with_time_or_count= s r c G s ddl m} || S )a The `with_latest_from` operator.
Merges the specified observable sequences into one observable
sequence by creating a tuple only when the first
observable sequence produces an element. The observables can be
passed either as separate arguments or as a list.
.. marble::
:alt: with_latest_from
---1---2---3----4-|
--a-----b----c-d----|
[with_latest_from() ]
---1,a-2,a-3,b--4,d-|
Examples:
>>> op = with_latest_from(obs1)
>>> op = with_latest_from([obs1, obs2, obs3])
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the result of
combining elements of the sources into a tuple.
r/ )with_latest_from_)Z_withlatestfromr )rX r r1 r1 r2 with_latest_fromG s r c G s ddl m} || S )a Merges the specified observable sequences into one observable
sequence by creating a tuple whenever all of the
observable sequences have produced an element at a corresponding
index.
.. marble::
:alt: zip
--1--2---3-----4---|
-a----b----c-d------|
[ zip() ]
--1,a-2,b--3,c-4,d-|
Example:
>>> res = zip(obs1, obs2)
Args:
args: Observable sources to zip.
Returns:
An operator function that takes an observable source and
returns an observable sequence containing the result of
combining elements of the sources as a tuple.
r/ )zip_)_zipr )rO r r1 r1 r2 zipg s r c C s ddl m} || S )a Merges the specified observable sequence and list into one
observable sequence by creating a tuple whenever all of
the observable sequences have produced an element at a
corresponding index.
.. marble::
:alt: zip_with_iterable
--1---2----3---4---|
[ zip(a,b,c,b) ]
--1,a-2,b--3,c-4,d-|
Example
>>> res = zip([1,2,3])
Args:
second: Iterable to zip with the source observable..
Returns:
An operator function that takes and observable source and
returns an observable sequence containing the result of
combining elements of the sources as a tuple.
r/ )zip_with_iterable_)r r )r r r1 r1 r2 zip_with_iterable s r throttle_with_timeoutto_list
zip_with_list)N)N)NN)N)N)N)N)N)N)NN)N)NN)NN)NNN)N)N)N)NN)N)N)N)N)N)N)N)N)N)N)NN)N)N)NN)N)N)N)N)N)N)N)N)N)N)NN)NN)NN)N)N)N)N)NN)FN)N)N)N)NNN)N)N)N)N)N)N)N)N)F)F)N)N)N)NN)NNN)N)N)N)r N)N)NN)N)asyncior r r r r r r r r r
r r r
r r r Z reactivexr r r r r r Zreactivex.internal.basicr Zreactivex.internal.utilsr Zreactivex.subjectr Zreactivex.typingr r r r r r r! r" r# r$ r% r& r' r( r) r* r+ r, boolr3 r6 r9 floatr< r@ rC rF r rJ Z RelativeTimeZ
SchedulerBaserO rQ ExceptionrT rW rZ ra re rG rk r ro ZAbsoluteOrRelativeTimers rw rz r| r~ r Z ObserverBaser ZOnNextZOnErrorZ OnCompletedr r r r r r r r Actionr r r r r r r r r r r r r r r r r r^ r r r r r_ r r r r Z SubjectBaser r r r r r strr r r r r r r r r r r! r$ r& r) r, rH r/ r2 r5 r8 r; r= r? rD rF rG rN rQ rS rT rX rY rZ r\ r^ r` rb rd rg rk rm ro rr ru rx rv ry r| r r r r r r r r r r r r r r r r r r r r r __all__r1 r1 r1 r2 s @$ & & " # % " + & ' & & ! # & " $ )$ #
'
%( *
" !$ &
$ ! "
#
%
-
-
-
8 % %
"&
* ! $
" #$
!
!" + ( # $ "" ! )
'
/
%
&$
$
' ( (" $$# ( '
# , %( &
"" )'((!,% %, $ " $ %
(
'
( (
$' % #
,