SWAMPy

✎ CARVE FLESH

EDITING: _groupjoin.py

import logging
from collections import OrderedDict
from typing import Any, Callable, Optional, Tuple, TypeVar

from reactivex import Observable, abc
from reactivex import operators as ops
from reactivex.disposable import (
    CompositeDisposable,
    RefCountDisposable,
    SingleAssignmentDisposable,
)
from reactivex.internal import add_ref
from reactivex.subject import Subject

_TLeft = TypeVar("_TLeft")
_TRight = TypeVar("_TRight")

log = logging.getLogger("Rx")

def group_join_(
    right: Observable[_TRight],
    left_duration_mapper: Callable[[_TLeft], Observable[Any]],
    right_duration_mapper: Callable[[_TRight], Observable[Any]],
) -> Callable[[Observable[_TLeft]], Observable[Tuple[_TLeft, Observable[_TRight]]]]:
    """Correlates the elements of two sequences based on overlapping
    durations, and groups the results.

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 observable sequence that contains elements combined into a tuple
    from source elements that have an overlapping duration.
    """

def nothing(_: Any) -> None:
        return None

def group_join(
        left: Observable[_TLeft],
    ) -> Observable[Tuple[_TLeft, Observable[_TRight]]]:
        def subscribe(
            observer: abc.ObserverBase[Tuple[_TLeft, Observable[_TRight]]],
            scheduler: Optional[abc.SchedulerBase] = None,
        ) -> abc.DisposableBase:
            group = CompositeDisposable()
            rcd = RefCountDisposable(group)
            left_map: OrderedDict[int, Subject[_TRight]] = OrderedDict()
            right_map: OrderedDict[int, _TRight] = OrderedDict()
            left_id = [0]
            right_id = [0]

def on_next_left(value: _TLeft) -> None:
                subject: Subject[_TRight] = Subject()

with left.lock:
                    _id = left_id[0]
                    left_id[0] += 1
                    left_map[_id] = subject

try:
                    result = (value, add_ref(subject, rcd))
                except Exception as e:
                    log.error("*** Exception: %s" % e)
                    for left_value in left_map.values():
                        left_value.on_error(e)

observer.on_error(e)
                    return

observer.on_next(result)

for right_value in right_map.values():
                    subject.on_next(right_value)

md = SingleAssignmentDisposable()
                group.add(md)

def expire():
                    if _id in left_map:
                        del left_map[_id]
                        subject.on_completed()

group.remove(md)

try:
                    duration = left_duration_mapper(value)
                except Exception as e:
                    for left_value in left_map.values():
                        left_value.on_error(e)

observer.on_error(e)
                    return

def on_error(error: Exception) -> Any:
                    for left_value in left_map.values():
                        left_value.on_error(error)

observer.on_error(error)

md.disposable = duration.pipe(ops.take(1)).subscribe(
                    nothing, on_error, expire, scheduler=scheduler
                )

def on_error_left(error: Exception) -> None:
                for left_value in left_map.values():
                    left_value.on_error(error)

observer.on_error(error)

group.add(
                left.subscribe(
                    on_next_left,
                    on_error_left,
                    observer.on_completed,
                    scheduler=scheduler,
                )
            )

def send_right(value: _TRight) -> None:
                with left.lock:
                    _id = right_id[0]
                    right_id[0] += 1
                    right_map[_id] = value

md = SingleAssignmentDisposable()
                group.add(md)

def expire():
                    del right_map[_id]
                    group.remove(md)

try:
                    duration = right_duration_mapper(value)
                except Exception as e:
                    for left_value in left_map.values():
                        left_value.on_error(e)

observer.on_error(e)
                    return

def on_error(error: Exception):
                    with left.lock:
                        for left_value in left_map.values():
                            left_value.on_error(error)

observer.on_error(error)

md.disposable = duration.pipe(ops.take(1)).subscribe(
                    nothing, on_error, expire, scheduler=scheduler
                )

with left.lock:
                    for left_value in left_map.values():
                        left_value.on_next(value)

def on_error_right(error: Exception) -> None:
                for left_value in left_map.values():
                    left_value.on_error(error)

observer.on_error(error)

group.add(right.subscribe(send_right, on_error_right, scheduler=scheduler))
            return rcd

return Observable(subscribe)

return group_join

__all__ = ["group_join_"]

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-02-16 09:47
📁 connectable/	DIR	—	2026-02-16 09:47
📄 __init__.py	PY	129.4 KB	2026-02-16 09:47	EDIT
📄 _all.py	PY	497 B	2026-02-16 09:47	EDIT
📄 _amb.py	PY	3 KB	2026-02-16 09:47	EDIT
📄 _asobservable.py	PY	821 B	2026-02-16 09:47	EDIT
📄 _average.py	PY	1.6 KB	2026-02-16 09:47	EDIT
📄 _buffer.py	PY	2.2 KB	2026-02-16 09:47	EDIT
📄 _bufferwithtime.py	PY	619 B	2026-02-16 09:47	EDIT
📄 _bufferwithtimeorcount.py	PY	557 B	2026-02-16 09:47	EDIT
📄 _catch.py	PY	2.5 KB	2026-02-16 09:47	EDIT
📄 _combinelatest.py	PY	798 B	2026-02-16 09:47	EDIT
📄 _concat.py	PY	695 B	2026-02-16 09:47	EDIT
📄 _contains.py	PY	581 B	2026-02-16 09:47	EDIT
📄 _count.py	PY	558 B	2026-02-16 09:47	EDIT
📄 _debounce.py	PY	5.4 KB	2026-02-16 09:47	EDIT
📄 _defaultifempty.py	PY	1.4 KB	2026-02-16 09:47	EDIT
📄 _delay.py	PY	4.6 KB	2026-02-16 09:47	EDIT
📄 _delaysubscription.py	PY	994 B	2026-02-16 09:47	EDIT
📄 _delaywithmapper.py	PY	3.4 KB	2026-02-16 09:47	EDIT
📄 _dematerialize.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _distinct.py	PY	2.5 KB	2026-02-16 09:47	EDIT
📄 _distinctuntilchanged.py	PY	2.8 KB	2026-02-16 09:47	EDIT
📄 _do.py	PY	10.5 KB	2026-02-16 09:47	EDIT
📄 _dowhile.py	PY	817 B	2026-02-16 09:47	EDIT
📄 _elementatordefault.py	PY	1.5 KB	2026-02-16 09:47	EDIT
📄 _exclusive.py	PY	2.4 KB	2026-02-16 09:47	EDIT
📄 _expand.py	PY	3 KB	2026-02-16 09:47	EDIT
📄 _filter.py	PY	2.7 KB	2026-02-16 09:47	EDIT
📄 _finallyaction.py	PY	1.3 KB	2026-02-16 09:47	EDIT
📄 _find.py	PY	1.4 KB	2026-02-16 09:47	EDIT
📄 _first.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _firstordefault.py	PY	2.5 KB	2026-02-16 09:47	EDIT
📄 _flatmap.py	PY	4.2 KB	2026-02-16 09:47	EDIT
📄 _forkjoin.py	PY	843 B	2026-02-16 09:47	EDIT
📄 _groupby.py	PY	852 B	2026-02-16 09:47	EDIT
📄 _groupbyuntil.py	PY	6.2 KB	2026-02-16 09:47	EDIT
📄 _groupjoin.py	PY	5.7 KB	2026-02-16 09:47	EDIT
📄 _ignoreelements.py	PY	929 B	2026-02-16 09:47	EDIT
📄 _isempty.py	PY	534 B	2026-02-16 09:47	EDIT
📄 _join.py	PY	4.5 KB	2026-02-16 09:47	EDIT
📄 _last.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _lastordefault.py	PY	1.9 KB	2026-02-16 09:47	EDIT
📄 _map.py	PY	2.1 KB	2026-02-16 09:47	EDIT
📄 _materialize.py	PY	1.4 KB	2026-02-16 09:47	EDIT
📄 _max.py	PY	995 B	2026-02-16 09:47	EDIT
📄 _maxby.py	PY	1018 B	2026-02-16 09:47	EDIT
📄 _merge.py	PY	5.1 KB	2026-02-16 09:47	EDIT
📄 _min.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _minby.py	PY	2.4 KB	2026-02-16 09:47	EDIT
📄 _multicast.py	PY	2.9 KB	2026-02-16 09:47	EDIT
📄 _observeon.py	PY	1.2 KB	2026-02-16 09:47	EDIT
📄 _onerrorresumenext.py	PY	419 B	2026-02-16 09:47	EDIT
📄 _pairwise.py	PY	1.6 KB	2026-02-16 09:47	EDIT
📄 _partition.py	PY	3 KB	2026-02-16 09:47	EDIT
📄 _pluck.py	PY	1.2 KB	2026-02-16 09:47	EDIT
📄 _publish.py	PY	2.2 KB	2026-02-16 09:47	EDIT
📄 _publishvalue.py	PY	901 B	2026-02-16 09:47	EDIT
📄 _reduce.py	PY	1.5 KB	2026-02-16 09:47	EDIT
📄 _repeat.py	PY	1 KB	2026-02-16 09:47	EDIT
📄 _replay.py	PY	2.4 KB	2026-02-16 09:47	EDIT
📄 _retry.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _sample.py	PY	2.1 KB	2026-02-16 09:47	EDIT
📄 _scan.py	PY	1.7 KB	2026-02-16 09:47	EDIT
📄 _sequenceequal.py	PY	4.1 KB	2026-02-16 09:47	EDIT
📄 _single.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _singleordefault.py	PY	2.8 KB	2026-02-16 09:47	EDIT
📄 _skip.py	PY	1.3 KB	2026-02-16 09:47	EDIT
📄 _skiplast.py	PY	1.6 KB	2026-02-16 09:47	EDIT
📄 _skiplastwithtime.py	PY	2.3 KB	2026-02-16 09:47	EDIT
📄 _skipuntil.py	PY	2.1 KB	2026-02-16 09:47	EDIT
📄 _skipuntilwithtime.py	PY	2.2 KB	2026-02-16 09:47	EDIT
📄 _skipwhile.py	PY	2.2 KB	2026-02-16 09:47	EDIT
📄 _skipwithtime.py	PY	2.2 KB	2026-02-16 09:47	EDIT
📄 _slice.py	PY	2.3 KB	2026-02-16 09:47	EDIT
📄 _some.py	PY	1.6 KB	2026-02-16 09:47	EDIT
📄 _startswith.py	PY	689 B	2026-02-16 09:47	EDIT
📄 _subscribeon.py	PY	1.7 KB	2026-02-16 09:47	EDIT
📄 _sum.py	PY	511 B	2026-02-16 09:47	EDIT
📄 _switchlatest.py	PY	2.8 KB	2026-02-16 09:47	EDIT
📄 _take.py	PY	1.4 KB	2026-02-16 09:47	EDIT
📄 _takelast.py	PY	1.6 KB	2026-02-16 09:47	EDIT
📄 _takelastbuffer.py	PY	1.7 KB	2026-02-16 09:47	EDIT
📄 _takelastwithtime.py	PY	2.3 KB	2026-02-16 09:47	EDIT
📄 _takeuntil.py	PY	1.5 KB	2026-02-16 09:47	EDIT
📄 _takeuntilwithtime.py	PY	1.7 KB	2026-02-16 09:47	EDIT
📄 _takewhile.py	PY	3.5 KB	2026-02-16 09:47	EDIT
📄 _takewithtime.py	PY	1.8 KB	2026-02-16 09:47	EDIT
📄 _throttlefirst.py	PY	1.8 KB	2026-02-16 09:47	EDIT
📄 _timeinterval.py	PY	1.5 KB	2026-02-16 09:47	EDIT
📄 _timeout.py	PY	3.2 KB	2026-02-16 09:47	EDIT
📄 _timeoutwithmapper.py	PY	4.3 KB	2026-02-16 09:47	EDIT
📄 _timestamp.py	PY	1.3 KB	2026-02-16 09:47	EDIT
📄 _todict.py	PY	1.9 KB	2026-02-16 09:47	EDIT
📄 _tofuture.py	PY	2 KB	2026-02-16 09:47	EDIT
📄 _toiterable.py	PY	1.1 KB	2026-02-16 09:47	EDIT
📄 _tomarbles.py	PY	2.3 KB	2026-02-16 09:47	EDIT
📄 _toset.py	PY	996 B	2026-02-16 09:47	EDIT
📄 _whiledo.py	PY	1 KB	2026-02-16 09:47	EDIT
📄 _window.py	PY	5.1 KB	2026-02-16 09:47	EDIT
📄 _windowwithcount.py	PY	2.6 KB	2026-02-16 09:47	EDIT
📄 _windowwithtime.py	PY	3.8 KB	2026-02-16 09:47	EDIT
📄 _windowwithtimeorcount.py	PY	3 KB	2026-02-16 09:47	EDIT
📄 _withlatestfrom.py	PY	898 B	2026-02-16 09:47	EDIT
📄 _zip.py	PY	2.3 KB	2026-02-16 09:47	EDIT