U
vi @ s d dl mZ d dlmZmZmZmZmZ d dlm Z m
Z
m Z d dlm Z
d dlmZmZmZ d dlmZ d dlmZ d dlmZ ed Zed
Zed Zdeeef eeeef ee eef ge
e f eeg ee f ee
e ge
e eef f d
ddZdgZ d S ) ) OrderedDict)AnyCallableOptionalTypeVarcast)GroupedObservable
Observableabc) operators)CompositeDisposableRefCountDisposableSingleAssignmentDisposable)identitySubject)Mapper_T_TKey_TValueN)
key_mapperelement_mapperduration_mappersubject_mapperreturnc sT |pt tttf tdd }|p$|tt ttttf d fdd }|S )aQ 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.
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.
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: 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.
c S s t S Nr r r R/opt/alt/python38/lib/python3.8/site-packages/reactivex/operators/_groupbyuntil.py< z!group_by_until_..)sourcer c s> dt jtttf tt j t jd fdd
}t|S )N)observer schedulerr c
s~ t t t td d f dd }td dfdd }d dfdd } j|||d
S ) N)xr c
sB d d z | W nJ t k
r^ } z, D ]}|| q.| W Y d S d }~X Y nX d} sz
W nJ t k
r } z, D ]}|| q| W Y d S d }~X Y nX < d}|rt }t}z |}W nN t k
rJ } z. D ]}|| q| W Y d S d }~X Y nX | t d d fdd td ddd}t d d fd
d }d d fd d
} | t
dj |||
d_
z | }
W nN t k
r2 } z. D ]}|| q | W Y d S d } ~ X Y nX |
d S )NFTr c s$ r= d S r ) on_completedremover )group_disposablekeysadwriterwritersr r expirez s z[group_by_until_..group_by_until..subscribe..on_next..expire)valuer c S s d S r r )r- r r r on_next s z\group_by_until_..group_by_until..subscribe..on_next..on_next)exnr c s& D ]}|| q | d S r valueson_error)r/ wrtr! r+ r r r2 s z]group_by_until_..group_by_until..subscribe..on_next..on_errorc s
d S r r r )r, r r r% s zagroup_by_until_..group_by_until..subscribe..on_next..on_completed r" ) Exceptionr1 r2 getr r. r addr pipeopsZtake subscribeZ
disposable) r# er3 Zfire_new_map_entrygroupZduration_groupdurationr. r2 r% elementerror) r element_mapper_r' r r! ref_count_disposabler" subject_mapper_r+ )r, r( r) r* r r. J sz
zKgroup_by_until_..group_by_until..subscribe..on_next)exr c s& D ]}|| q | d S r r0 )rE r3 r4 r r r2 s zLgroup_by_until_..group_by_until..subscribe..on_errorr$ c s" D ] } | q d S r )r1 r% )r3 r4 r r r% s
zPgroup_by_until_..group_by_until..subscribe..on_completedr6 )r r r
r r7 r9 r< )r! r" r. r2 r% )r rB r r rD )r' r! rC r" r+ r r< B s $Qz:group_by_until_..group_by_until..subscribe)N) r
Z ObserverBaser r r r Z
SchedulerBaseZDisposableBaser )r r< r rB r rD )r r group_by_until? s jz'group_by_until_..group_by_until)r r r r r r r r )r r r r Zdefault_subject_mapperrG r rF r group_by_until_ s &orH )N) collectionsr typingr r r r r Z reactivexr r r
r r; Zreactivex.disposabler r
r Zreactivex.internal.basicr Zreactivex.subjectr Zreactivex.typingr r r r rH __all__r r r r s(