ReactiveX / RxPY / 30341405946

94%

master: 94%

Pull #812

github

fix(operators): synchronize window_with_time_or_count

The window state (`n`, `s`, `window_id`) was mutated from both the source
thread (`on_next`) and the scheduler thread (the timer `action`) with no
synchronization.

When the timer closes a window it calls `s.on_completed()` before
reassigning `s = Subject()`. `on_completed()` synchronously runs the whole
downstream chain, so while any user work is in progress the source thread
keeps calling `s.on_next(x)` on a subject that has already completed, and
those items are silently discarded.

Guard `action`, `on_next`, `on_error` and `on_completed` with
`synchronized(source.lock)`, matching what `window_with_time` already does
and what Rx.NET's `WindowTimeCount` does with its gate. `buffer_with_time_
or_count` is built on this operator and was affected too.

Fixes #694

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

24 of 24 new or added lines in 2 files covered. (100.0%)

25736 of 27525 relevant lines covered (93.5%)

0.94 hits per line