U
vi, @ s d dl mZ d dlZd dlZd dlmZmZmZmZm Z m
Z
m Z m Z d dl
mZ d dlmZ d dlmZ d dlmZ dd lmZ e d
Ze d Ze d Ze d
Ze dZe dZ e dZ e dddZ G dd dej e Z dgZ!dS ) ) annotationsN)AnyCallable GeneratorOptionalTypeVarUnioncastoverload)abc)
Disposable)CurrentThreadScheduler)AsyncIOScheduler )AutoDetachObserver_A_B_C_D_E_F_G_T_outT) covariantc
sn e Zd ZdZdGddd fdd
ZdHdd d
d d d
ZdIdddddd d
dddZedddddZedddd d dZeddd d d d!dZeddd d"d#d$d%dZeddd d"d&d'd(d)dZeddd d"d&d*d+d,d-dZeddd d"d&d*d.d/d0d1dZd2d3d4d5dZd3d6d7d8Z d9d6d:d;Z
dd?Z d` into
:mod:`operators `.Nz"Optional[abc.Subscription[_T_out]]None) subscribereturnc s t t | _|| _dS )zCreates an observable sequence object from the specified
subscription function.
Args:
subscribe: [Optional] Subscription function
N)super__init__ threadingRLocklock
_subscribe)selfr __class__ P/opt/alt/python38/lib/python3.8/site-packages/reactivex/observable/observable.pyr s
zObservable.__init__zabc.ObserverBase[_T_out]Optional[abc.SchedulerBase]abc.DisposableBase)observer schedulerr c C s | j r| ||S t S N)r# r )r$ r+ r, r' r' r( _subscribe_core, s zObservable._subscribe_corer, zCOptional[Union[abc.ObserverBase[_T_out], abc.OnNext[_T_out], None]]zOptional[abc.OnError]zOptional[abc.OnCompleted])on_nexton_error on_completedr, r c s t |tjs$t|drFtt|drFttjt |}|j}|j }|j
}t ||| dddddddd d
d fd d
}t
}| r|| n| t jS )a Subscribe an observer to the observable sequence.
You may subscribe using an observer or callbacks, not both; if the first
argument is an instance of :class:`Observer <..abc.ObserverBase>` or if
it has a (callable) attribute named :code:`on_next`, then any callback
arguments will be ignored.
Examples:
>>> source.subscribe()
>>> source.subscribe(observer)
>>> source.subscribe(observer, scheduler=scheduler)
>>> source.subscribe(on_next)
>>> source.subscribe(on_next, on_error)
>>> source.subscribe(on_next, on_error, on_completed)
>>> source.subscribe(on_next, on_error, on_completed, scheduler=scheduler)
Args:
observer: [Optional] The object that is to receive
notifications.
on_error: [Optional] Action to invoke upon exceptional termination
of the observable sequence.
on_completed: [Optional] Action to invoke upon graceful termination
of the observable sequence.
on_next: [Optional] Action to invoke for each element in the
observable sequence.
scheduler: [Optional] The default scheduler to use for this
subscription.
Returns:
Disposable object representing an observer's subscription to
the observable sequence.
r0 z-Union[abc.DisposableBase, Callable[[], None]]r* )
subscriberr c S s* t | tjst| dr"ttj| S t| S )zgFixes subscriber to make sure it returns a Disposable instead
of None or a dispose functiondispose)
isinstancer ZDisposableBasehasattrr r )r3 r' r' r( fix_subscriberk s z,Observable.subscribe..fix_subscriberNr) r r )___r c
sN z }W n. tk
r> } z |s. W 5 d }~X Y n X | _d S r- )r. ExceptionZfailZ subscription)r8 r9 r3 exZauto_detach_observerr7 r, r$ r' r( set_disposabley s
z,Observable.subscribe..set_disposable)NN)r5 r Z ObserverBaser6 callablegetattrr r r0 r1 r2 r r
Z singletonZschedule_requiredZscheduler r4 )r$ r0 r1 r2 r, Zobvr= Zcurrent_thread_schedulerr' r< r( r 3 s0 +
zObservable.subscribez"Callable[[Observable[_T_out]], _A]r )_Observable__op1r c C s d S r- r' )r$ r@ r' r' r( pipe s zObservable.pipezCallable[[_A], _B]r )r@ _Observable__op2r c C s d S r- r' )r$ r@ rB r' r' r( rA s zCallable[[_B], _C]r )r@ rB _Observable__op3r c C s d S r- r' )r$ r@ rB rC r' r' r( rA s zCallable[[_C], _D]r )r@ rB rC _Observable__op4r c C s d S r- r' )r$ r@ rB rC rD r' r' r( rA s zCallable[[_D], _E]r )r@ rB rC rD _Observable__op5r c C s d S r- r' )r$ r@ rB rC rD rE r' r' r( rA s zCallable[[_E], _F]r )r@ rB rC rD rE _Observable__op6r c C s d S r- r' )r$ r@ rB rC rD rE rF r' r' r( rA s zCallable[[_F], _G]r )r@ rB rC rD rE rF _Observable__op7r c C s d S r- r' )r$ r@ rB rC rD rE rF rG r' r' r( rA s
zCallable[[Any], Any]r ) operatorsr c G s ddl m } || f| S )a? Compose multiple operators left to right.
Composes zero or more operators into a functional composition.
The operators are composed from left to right. A composition of zero
operators gives back the original source.
Examples:
>>> source.pipe() == source
>>> source.pipe(f) == f(source)
>>> source.pipe(g, f) == f(g(source))
>>> source.pipe(h, g, f) == f(g(h(source)))
Args:
operators: Sequence of operators.
Returns:
The composed observable.
r rA rI )r$ rH Zpipe_r' r' r( rA s )r c C s ddl m } || S )a] Run source synchronously.
Subscribes to the observable source. Then blocks and waits for the
observable source to either complete or error. Returns the
last value emitted, or throws exception if any error occurred.
Examples:
>>> result = run(source)
Raises:
SequenceContainsNoElementsError: if observable completes
(on_completed) without any values being emitted.
Exception: raises exception if any error (on_error) occurred.
Returns:
The last element emitted from the observable.
r runrJ )r$ rK r' r' r( rK s zObservable.runz Generator[Any, None, _T_out]c C s2 ddl m} t }| |t|dd}| S )zmAwaits the given observable.
Returns:
The last item of the observable sequence.
r )
to_future_)loopr/ )Zoperators._tofuturerL asyncioget_event_looprA r __await__)r$ rL rM futurer' r' r( rP s zObservable.__await__zObservable[_T_out])otherr c C s ddl m} || |S )a Pythonic version of :func:`concat `.
Example:
>>> zs = xs + ys
Args:
other: The second observable sequence in the concatenation.
Returns:
Concatenated observable sequence.
r concat reactivexrT r$ rR rT r' r' r( __add__ s zObservable.__add__z'Observable[_T_out]'c C s ddl m} || |S )zPythonic use of :func:`concat `.
Example:
>>> xs += ys
Args:
other: The second observable sequence in the concatenation.
Returns:
Concatenated observable sequence.
r rS rU rW r' r' r( __iadd__ s zObservable.__iadd__zUnion[slice, int])keyr c C sR t |tr"|j|j|j }}}n||d d }}}ddlm} ||||| S )a$
Pythonic version of :func:`slice `.
Slices the given observable using Python slice notation. The arguments
to slice are `start`, `stop` and `step` given within brackets `[]` and
separated by the colons `:`.
It is basically a wrapper around the operators
:func:`skip `,
:func:`skip_last `,
:func:`take `,
:func:`take_last ` and
:func:`filter `.
The following diagram helps you remember how slices works with streams.
Positive numbers are relative to the start of the events, while negative
numbers are relative to the end (close) of the stream.
.. code::
r---e---a---c---t---i---v---e---!
0 1 2 3 4 5 6 7 8
-8 -7 -6 -5 -4 -3 -2 -1 0
Examples:
>>> result = source[1:10]
>>> result = source[1:-2]
>>> result = source[1:-1:2]
Args:
key: Slice object
Returns:
Sliced observable sequence.
Raises:
TypeError: If key is not of type :code:`int` or :code:`slice`
r )slice_)r5 slicestartstopstepZoperators._slicer\ )r$ rZ r^ r_ r` r\ r' r' r( __getitem__, s
(
zObservable.__getitem__)N)N)NNN)__name__
__module__ __qualname____doc__r r. r r
rA rK rP rX rY ra
__classcell__r' r' r% r( r s< b
r )"
__future__r rN r typingr r r r r r r r
rV r Zreactivex.disposabler Zreactivex.schedulerr
Z reactivex.scheduler.eventloopr r+ r r r r r r r r r ZObservableBaser __all__r' r' r' r( s( ( F