Merge 038e894 into 964fcde

ReactiveX · Mar 12, 2022 · d21978e · d21978e
2 parents 964fcde + 038e894
commit d21978e
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 6 deletions.
diff --git a/reactivex/observable/withlatestfrom.py b/reactivex/observable/withlatestfrom.py
@@ -41,15 +41,14 @@ def on_next(value: Any) -> None:
                         result = (value,) + tuple(values)
                         observer.on_next(result)
 
+            children_subscription = [
+                subscribechild(i, child) for i, child in enumerate(children)
+            ]
             disp = parent.subscribe(
                 on_next, observer.on_error, observer.on_completed, scheduler=scheduler
             )
             parent_subscription.disposable = disp
 
-            children_subscription = [
-                subscribechild(i, child) for i, child in enumerate(children)
-            ]
-
             return [parent_subscription] + children_subscription
 
         return CompositeDisposable(subscribeall(parent, *sources))

diff --git a/reactivex/operators/__init__.py b/reactivex/operators/__init__.py
@@ -2427,10 +2427,32 @@ def repeat(
     return repeat_(repeat_count)
 
 
+@overload
+def replay(
+    buffer_size: Optional[int] = None,
+    window: Optional[typing.RelativeTime] = None,
+    *,
+    scheduler: Optional[abc.SchedulerBase] = None,
+) -> Callable[[Observable[_T1]], ConnectableObservable[_T1]]:
+    ...
+
+
+@overload
 def replay(
-    mapper: Optional[Mapper[Observable[_T1], Observable[_T2]]] = None,
     buffer_size: Optional[int] = None,
     window: Optional[typing.RelativeTime] = None,
+    *,
+    mapper: Optional[Mapper[Observable[_T1], Observable[_T2]]],
+    scheduler: Optional[abc.SchedulerBase] = None,
+) -> Callable[[Observable[_T1]], Observable[_T2]]:
+    ...
+
+
+def replay(
+    buffer_size: Optional[int] = None,
+    window: Optional[typing.RelativeTime] = None,
+    *,
+    mapper: Optional[Mapper[Observable[_T1], Observable[_T2]]] = None,
     scheduler: Optional[abc.SchedulerBase] = None,
 ) -> Callable[[Observable[_T1]], Union[Observable[_T2], ConnectableObservable[_T1]]]:
     """The `replay` operator.

diff --git a/reactivex/operators/_replay.py b/reactivex/operators/_replay.py
@@ -58,7 +58,7 @@ def subject_factory(
 
         return ops.multicast(subject_factory=subject_factory, mapper=mapper)
     rs: ReplaySubject[_TSource] = ReplaySubject(buffer_size, window, scheduler)
-    return ops.multicast(rs)
+    return ops.multicast(subject=rs)
 
 
 __all__ = ["replay_"]