Use synchronized wrapper instead

ReactiveX · Mar 12, 2022 · 3e2ef0c · 3e2ef0c
1 parent 9c344f1
commit 3e2ef0c
Show file tree

Hide file tree

Showing 10 changed files with 30 additions and 25 deletions.
diff --git a/reactivex/internal/__init__.py b/reactivex/internal/__init__.py
@@ -7,13 +7,18 @@
     SequenceContainsNoElementsError,
 )
 from .priorityqueue import PriorityQueue
+from .utils import NotSet, add_ref, alias, infinite
 
 __all__ = [
+    "add_ref",
+    "alias",
     "ArgumentOutOfRangeException",
     "DisposedException",
     "default_comparer",
     "default_error",
+    "infinite",
     "noop",
+    "NotSet",
     "SequenceContainsNoElementsError",
     "concurrency",
     "DELTA_ZERO",

diff --git a/reactivex/internal/utils.py b/reactivex/internal/utils.py
@@ -53,3 +53,6 @@ def __eq__(self, other: Any) -> bool:
 
     def __repr__(self) -> str:
         return "NotSet"
+
+
+__all__ = ["add_ref", "infinite", "alias", "NotSet"]
diff --git a/reactivex/observable/zip.py b/reactivex/observable/zip.py
@@ -4,7 +4,7 @@
 
 from reactivex import Observable, abc, from_future
 from reactivex.disposable import CompositeDisposable, SingleAssignmentDisposable
-from reactivex.internal.concurrency import synchronized
+from reactivex.internal import synchronized
 
 
 def zip_(*args: Observable[Any]) -> Observable[Tuple[Any, ...]]:

diff --git a/reactivex/operators/_groupjoin.py b/reactivex/operators/_groupjoin.py
@@ -9,7 +9,7 @@
     RefCountDisposable,
     SingleAssignmentDisposable,
 )
-from reactivex.internal.utils import add_ref
+from reactivex.internal import add_ref
 from reactivex.subject import Subject
 
 _TLeft = TypeVar("_TLeft")

diff --git a/reactivex/operators/_merge.py b/reactivex/operators/_merge.py
@@ -4,7 +4,7 @@
 import reactivex
 from reactivex import Observable, abc, from_future, typing
 from reactivex.disposable import CompositeDisposable, SingleAssignmentDisposable
-from reactivex.internal.concurrency import synchronized
+from reactivex.internal import synchronized
 
 _T = TypeVar("_T")
 

diff --git a/reactivex/operators/_window.py b/reactivex/operators/_window.py
@@ -9,8 +9,7 @@
     SerialDisposable,
     SingleAssignmentDisposable,
 )
-from reactivex.internal import noop
-from reactivex.internal.utils import add_ref
+from reactivex.internal import add_ref, noop
 from reactivex.subject import Subject
 
 log = logging.getLogger("Rx")

diff --git a/reactivex/operators/_windowwithcount.py b/reactivex/operators/_windowwithcount.py
@@ -3,8 +3,7 @@
 
 from reactivex import Observable, abc
 from reactivex.disposable import RefCountDisposable, SingleAssignmentDisposable
-from reactivex.internal.exceptions import ArgumentOutOfRangeException
-from reactivex.internal.utils import add_ref
+from reactivex.internal import ArgumentOutOfRangeException, add_ref
 from reactivex.subject import Subject
 
 log = logging.getLogger("Rx")

diff --git a/reactivex/operators/_windowwithtime.py b/reactivex/operators/_windowwithtime.py
@@ -8,8 +8,7 @@
     SerialDisposable,
     SingleAssignmentDisposable,
 )
-from reactivex.internal.constants import DELTA_ZERO
-from reactivex.internal.utils import add_ref
+from reactivex.internal import DELTA_ZERO, add_ref, synchronized
 from reactivex.scheduler import TimeoutScheduler
 from reactivex.subject import Subject
 
@@ -69,18 +68,18 @@ def create_timer():
                 if is_shift:
                     next_shift[0] += timeshift
 
+                @synchronized(source.lock)
                 def action(scheduler: abc.SchedulerBase, state: Any = None):
                     s: Optional[Subject[_T]] = None
 
-                    with source.lock:
-                        if is_shift:
-                            s = Subject()
-                            queue.append(s)
-                            observer.on_next(add_ref(s, ref_count_disposable))
+                    if is_shift:
+                        s = Subject()
+                        queue.append(s)
+                        observer.on_next(add_ref(s, ref_count_disposable))
 
-                        if is_span:
-                            s = queue.pop(0)
-                            s.on_completed()
+                    if is_span:
+                        s = queue.pop(0)
+                        s.on_completed()
 
                     create_timer()
 
@@ -95,17 +94,17 @@ def on_next(x: _T) -> None:
                     for s in queue:
                         s.on_next(x)
 
+            @synchronized(source.lock)
             def on_error(e: Exception) -> None:
-                with source.lock:
-                    for s in queue:
-                        s.on_error(e)
+                for s in queue:
+                    s.on_error(e)
 
                 observer.on_error(e)
 
+            @synchronized(source.lock)
             def on_completed() -> None:
-                with source.lock:
-                    for s in queue:
-                        s.on_completed()
+                for s in queue:
+                    s.on_completed()
 
                 observer.on_completed()
 

diff --git a/reactivex/operators/_windowwithtimeorcount.py b/reactivex/operators/_windowwithtimeorcount.py
@@ -7,7 +7,7 @@
     SerialDisposable,
     SingleAssignmentDisposable,
 )
-from reactivex.internal.utils import add_ref
+from reactivex.internal import add_ref
 from reactivex.scheduler import TimeoutScheduler
 from reactivex.subject import Subject
 

diff --git a/tests/test_observable/test_withlatestfrom.py b/tests/test_observable/test_withlatestfrom.py
@@ -20,7 +20,7 @@ class RxException(Exception):
 # Helper function for raising exceptions within lambdas
 
 
-def _raise(ex):
+def _raise(ex: str) -> None:
     raise RxException(ex)