U
vi @ s d dl mZ d dlmZ d dlmZmZmZmZ d dl m
Z
m Z m Z d dl
mZmZ d dlmZ e
e e
eedf dd d
Zd
gZd S ) )Future)RLock)AnyListOptionalTuple)
Observableabc from_future)CompositeDisposableSingleAssignmentDisposable) synchronized.)argsreturnc s4 t | dtjt ttj td fdd
}t|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.
Example:
>>> res = zip(obs1, obs2)
Args:
args: Observable sources to zip.
Returns:
An observable sequence containing the result of combining
elements of the sources as tuple.
N)observer schedulerr c s t }dd t|D t }dg| t|td dfdd td dfdd d g| td d fd d
}t|D ] }|| qtS ) Nc S s g | ]}g qS r ).0_r r I/opt/alt/python38/lib/python3.8/site-packages/reactivex/observable/zip.py
! s z+zip_..subscribe..F)ir c
s t dd D rzdd D }t|}W n2 tk
r^ } z| W Y d S d }~X Y nX | tdd t D r d S )Nc s s | ]}t |V qd S Nlen)r qr r r ' s z9zip_..subscribe..next_..c S s g | ]}| d qS )r )pop)r xr r r r ) s z:zip_..subscribe..next_..c s s" | ]\}}t |d kr|V qdS )r Nr )r queuedoner r r r 4 s )alltuple Exceptionon_erroron_nextanyzip on_completed)r Z
queued_valuesresex is_completedr queuesr r next_% s
z&zip_..subscribe..next_c s$ d | < t | dkr d S )NTr )r r( r r+ r r completed< s z*zip_..subscribe..completedc sd }t |trt|}t }td d fdd }|j|j fddd|_| < d S )N)r r c s | d S r )append)r )r r. r- r r r% J s z6zip_..subscribe..func..on_nextc s S r r r )r0 r r r O z7zip_..subscribe..func..)r )
isinstancer r
r r subscriber$ Z
disposable)r sourceZsadr% )r0 r. r r- r sources
subscriptionsr/ r funcC s
z%zip_..subscribe..func)r ranger r
intr )r r nlockr9 idxr7 )r0 r, r. r r- r r8 r r5 s
zzip_..subscribe)N)listr Z ObserverBaser r Z
SchedulerBaser r )r r5 r r? r zip_
s :rA N)asyncior threadingr typingr r r r Z reactivexr r r
Zreactivex.disposabler r Zreactivex.internalr
rA __all__r r r r s P