Repository: ReactiveX/RxPY

Issue:
Add forkJoin Operator
Hi y'all, it's been great using RxPy, but I noticed that [`forkJoin`](https://www.learnrxjs.io/learn-rxjs/operators/combination/forkjoin) was missing from the list of operators. I needed this for a project recently where I wanted a bunch of network requests to happen concurrently, but I didn't want to the stream to emit until they all finished. Here's what I did with `asyncio` to accomplish similar functionality:

```py
import asyncio
import rx
import rx.operators as ops
import random

from rx.scheduler.eventloop import AsyncIOScheduler

index = 0


async def request():
    global index
    duration = random.randint(1, 4)
    request_id = index
    index += 1
    print(f'request {request_id} will take {duration} s')
    await asyncio.sleep(duration)
    print(f'request {request_id} done!')
    return request_id


def schedule_requests(_):
    tasks = [asyncio.create_task(request()) for _ in range(5)]
    return rx.from_future(asyncio.gather(*tasks))


async def main():
    done = asyncio.Future()

    rx.of('cool beans').pipe(
      ops.flat_map(schedule_requests),
    ).subscribe(
      on_next=print,
      scheduler=AsyncIOScheduler(asyncio.get_running_loop()),
    )

    await done

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
```

Still, it would be lovely if `forkJoin` could be added and have a true Rx implementation. The code for [`forkJoin` in TypeScript](https://github.com/ReactiveX/rxjs/blob/edd67313814bfc32e8a5129d8049e4d4678cd35d/src/internal/observable/forkJoin.ts#L141-L203) looks relatively simple enough to port over to Python 

Discussion:
coveralls: 
[![Coverage Status](https://coveralls.io/builds/32631505/badge)](https://coveralls.io/builds/32631505)

Coverage increased (+0.03%) to 92.495% when pulling **3ca0b574c255c3bfa65f2118dfcc4a0895b46592 on hoc081098:hoc081098/forkjoin** into **f2642a87e6b4d5b3dc9a482323716cf15a6ef570 on ReactiveX:master**.


Patch:
diff --git a/docs/operators.rst b/docs/operators.rst
index fa7215ee..88083cfa 100644
--- a/docs/operators.rst
+++ b/docs/operators.rst
@@ -69,6 +69,7 @@ Operator                                                                    Desc
 :func:`start_with <rx.operators.start_with>`            Emit a specified sequence of items before beginning to emit the items from the source Observable.
 :func:`switch_latest <rx.operators.switch_latest>`      Convert an Observable that emits Observables into a single Observable that emits the items emitted by the most-recently-emitted of those Observables.
 :func:`zip <rx.operators.zip>`                          Combine the emissions of multiple Observables together via a specified function and emit single items for each combination based on the results of this function.
+:func:`fork_join <rx.operators.fork_join>`              Wait for Observables to complete and then combine last values they emitted into a tuple.
 ======================================================  ================================================
 
 Error Handling
diff --git a/rx/__init__.py b/rx/__init__.py
index b995f2b2..9ba47010 100644
--- a/rx/__init__.py
+++ b/rx/__init__.py
@@ -6,7 +6,6 @@ from typing import Iterable, Callable, Any, Optional, Union, Mapping
 from .core import Observable, pipe, typing
 from .internal.utils import alias
 
-
 # Please make sure the version here remains the same as in project.cfg
 __version__ = '3.1.1'
 
@@ -319,6 +318,35 @@ def for_in(values: Iterable[Any], mapper: typing.Mapper) -> Observable:
     return concat_with_iterable(map(mapper, values))
 
 
+def fork_join(*sources: Observable) -> Observable:
+    """Wait for observables to complete and then combine last values
+    they emitted into a tuple. Whenever any of that observables completes
+    without emitting any value, result sequence will complete at that moment as well.
+
+    .. marble::
+        :alt: fork_join
+
+        ---a-----b--c---d-|
+        --1---2------3-4---|
+        -a--------- b---|
+        [      fork_join()     ]
+        --------------------d4b|
+
+    Examples:
+        >>> obs = rx.fork_join(obs1, obs2, obs3)
+
+    Args:
+        sources: Sequence of observables.
+
+    Returns:
+        An observable sequence containing the result of combining last element from
+        each source in given sequence.
+    """
+
+    from .core.observable.forkjoin import _fork_join
+    return _fork_join(*sources)
+
+
 def from_callable(supplier: Callable[[], Any],
                   scheduler: Optional[typing.Scheduler] = None
                   ) -> Observable:
@@ -561,8 +589,8 @@ def generate(initial_state: Any,
 
 
 def hot(string: str,
-        timespan: typing.RelativeTime=0.1,
-        duetime:typing.AbsoluteOrRelativeTime = 0.0,
+        timespan: typing.RelativeTime = 0.1,
+        duetime: typing.AbsoluteOrRelativeTime = 0.0,
         scheduler: Optional[typing.Scheduler] = None,
         lookup: Optional[Mapping] = None,
         error: Optional[Exception] = None
@@ -1092,8 +1120,8 @@ def zip(*args: Observable) -> Observable:
         :alt: zip
 
         --1--2---3-----4---|
-        -a----b----c-d-----|
-        [       zip()      ]
+        -a----b----c-d------|
+        [       zip()       ]
         --1,a-2,b--3,c-4,d-|
 
     Example:
diff --git a/rx/core/observable/forkjoin.py b/rx/core/observable/forkjoin.py
new file mode 100644
index 00000000..2b324375
--- /dev/null
+++ b/rx/core/observable/forkjoin.py
@@ -0,0 +1,69 @@
+from typing import Optional
+
+from rx.core import Observable, typing
+from rx.disposable import CompositeDisposable, SingleAssignmentDisposable
+
+
+def _fork_join(*sources: Observable) -> Observable:
+    """Wait for observables to complete and then combine last values
+    they emitted into a tuple. Whenever any of that observables completes
+    without emitting any value, result sequence will complete at that moment as well.
+
+    Examples:
+        >>> obs = rx.fork_join(obs1, obs2, obs3)
+
+    Returns:
+        An observable sequence containing the result of combining last element from
+        each source in given sequence.
+    """
+
+    parent = sources[0]
+
+    def subscribe(observer: typing.Observer,
+                  scheduler: Optional[typing.Scheduler] = None
+                  ) -> CompositeDisposable:
+        n = len(sources)
+        values = [None] * n
+        is_done = [False] * n
+        has_value = [False] * n
+
+        def done(i: int):
+            is_done[i] = True
+
+            if not has_value[i]:
+                observer.on_completed()
+                return
+
+            if all(is_done):
+                if all(has_value):
+                    observer.on_next(tuple(values))
+                    observer.on_completed()
+                else:
+                    observer.on_completed()
+
+        subscriptions = [None] * n
+
+        def _subscribe(i: int):
+            subscriptions[i] = SingleAssignmentDisposable()
+
+            def on_next(value):
+                with parent.lock:
+                    values[i] = value
+                    has_value[i] = True
+
+            def on_completed():
+                with parent.lock:
+                    done(i)
+
+            subscriptions[i].disposable = sources[i].subscribe_(
+                on_next,
+                observer.on_error,
+                on_completed,
+                scheduler
+            )
+
+        for i in range(n):
+            _subscribe(i)
+        return CompositeDisposable(subscriptions)
+
+    return Observable(subscribe)
diff --git a/rx/core/observable/zip.py b/rx/core/observable/zip.py
index 530ab8c8..8b756477 100644
--- a/rx/core/observable/zip.py
+++ b/rx/core/observable/zip.py
@@ -1,11 +1,11 @@
+from threading import RLock
 from typing import Optional, List
 
 from rx import from_future
 from rx.core import Observable, typing
 from rx.disposable import CompositeDisposable, SingleAssignmentDisposable
-from rx.internal.utils import is_future
 from rx.internal.concurrency import synchronized
-from threading import RLock
+from rx.internal.utils import is_future
 
 
 # pylint: disable=redefined-builtin
@@ -29,11 +29,11 @@ def _zip(*args: Observable) -> Observable:
 
     sources = list(args)
 
-    def subscribe(observer: typing.Observer, scheduler: Optional[typing.Scheduler] = None):
+    def subscribe(observer: typing.Observer,
+                  scheduler: Optional[typing.Scheduler] = None) -> CompositeDisposable:
         n = len(sources)
         queues: List[List] = [[] for _ in range(n)]
         lock = RLock()
-        is_done = [False] * n
 
         @synchronized(lock)
         def next(i):
@@ -46,13 +46,6 @@ def _zip(*args: Observable) -> Observable:
                     return
 
                 observer.on_next(res)
-            elif all([x for j, x in enumerate(is_done) if j != i]):
-                observer.on_completed()
-
-        def done(i):
-            is_done[i] = True
-            if all(is_done):
-                observer.on_completed()
 
         subscriptions = [None] * n
 
@@ -65,7 +58,7 @@ def _zip(*args: Observable) -> Observable:
                 queues[i].append(x)
                 next(i)
 
-            sad.disposable = source.subscribe_(on_next, observer.on_error, lambda: done(i), scheduler)
+            sad.disposable = source.subscribe_(on_next, observer.on_error, observer.on_completed, scheduler)
             subscriptions[i] = sad
 
         for idx in range(n):
diff --git a/rx/core/operators/forkjoin.py b/rx/core/operators/forkjoin.py
new file mode 100644
index 00000000..8ce3ca90
--- /dev/null
+++ b/rx/core/operators/forkjoin.py
@@ -0,0 +1,22 @@
+from typing import Callable
+
+import rx
+from rx import Observable
+
+
+def _fork_join(*args: Observable) -> Callable[[Observable], Observable]:
+    def fork_join(source: Observable) -> Observable:
+        """Wait for observables to complete and then combine last values
+        they emitted into a tuple. Whenever any of that observables completes
+        without emitting any value, result sequence will complete at that moment as well.
+
+        Examples:
+            >>> obs = fork_join(source)
+
+        Returns:
+            An observable sequence containing the result of combining last element from
+            each source in given sequence.
+        """
+        return rx.fork_join(source, *args)
+
+    return fork_join
diff --git a/rx/core/operators/tofuture.py b/rx/core/operators/tofuture.py
index 67197888..f0f1d435 100644
--- a/rx/core/operators/tofuture.py
+++ b/rx/core/operators/tofuture.py
@@ -38,18 +38,20 @@ def _to_future(future_ctor: Optional[Callable[[], Future]] = None,
             has_value = True
 
         def on_error(err):
-            future.set_exception(err)
+            if not future.cancelled():
+                future.set_exception(err)
 
         def on_completed():
             nonlocal last_value
-            if has_value:
-                future.set_result(last_value)
-            else:
-                future.set_exception(SequenceContainsNoElementsError())
+            if not future.cancelled():
+                if has_value:
+                    future.set_result(last_value)
+                else:
+                    future.set_exception(SequenceContainsNoElementsError())
             last_value = None
 
-        source.subscribe_(on_next, on_error, on_completed, scheduler=scheduler)
+        dis = source.subscribe_(on_next, on_error, on_completed, scheduler=scheduler)
+        future.add_done_callback(lambda _: dis.dispose())
 
-        # No cancellation can be done
         return future
     return to_future
diff --git a/rx/operators/__init__.py b/rx/operators/__init__.py
index c2373d76..cdbd122e 100644
--- a/rx/operators/__init__.py
+++ b/rx/operators/__init__.py
@@ -1178,6 +1178,33 @@ def flat_map_latest(mapper: Mapper) -> Callable[[Observable], Observable]:
     return _flat_map_latest(mapper)
 
 
+def fork_join(*others: Observable) -> Callable[[Observable], Observable]:
+    """Wait for observables to complete and then combine last values
+    they emitted into a tuple. Whenever any of that observables completes
+    without emitting any value, result sequence will complete at that moment as well.
+
+    .. marble::
+        :alt: fork_join
+
+        ---a-----b--c---d-|
+        --1---2------3-4---|
+        -a--------- b---|
+        [      fork_join()     ]
+        --------------------d4b|
+
+    Examples:
+        >>> res = fork_join(obs1)
+        >>> res = fork_join(obs1, obs2, obs3)
+
+    Returns:
+        An operator function that takes an observable source and
+        return an observable sequence containing the result
+        of combining last element from each source in given sequence.
+    """
+    from rx.core.operators.forkjoin import _fork_join
+    return _fork_join(*others)
+
+
 def group_by(key_mapper: Mapper,
              element_mapper: Optional[Mapper] = None,
              subject_mapper: Optional[Callable[[], Subject]] = None,
@@ -3492,8 +3519,8 @@ def zip(*args: Observable) -> Callable[[Observable], Observable]:
         :alt: zip
 
         --1--2---3-----4---|
-        -a----b----c-d-----|
-        [       zip()      ]
+        -a----b----c-d------|
+        [       zip()       ]
         --1,a-2,b--3,c-4,d-|
 
 


Test patch:
diff --git a/tests/test_observable/test_forkjoin.py b/tests/test_observable/test_forkjoin.py
new file mode 100644
index 00000000..ca192df2
--- /dev/null
+++ b/tests/test_observable/test_forkjoin.py
@@ -0,0 +1,205 @@
+import unittest
+
+import rx
+from rx import operators as ops
+from rx.testing import ReactiveTest, TestScheduler
+
+on_next = ReactiveTest.on_next
+on_completed = ReactiveTest.on_completed
+on_error = ReactiveTest.on_error
+subscribe = ReactiveTest.subscribe
+subscribed = ReactiveTest.subscribed
+disposed = ReactiveTest.disposed
+created = ReactiveTest.created
+
+
+class RxException(Exception):
+    pass
+
+
+class TestForkJoin(unittest.TestCase):
+    def test_fork_join_never_never(self):
+        scheduler = TestScheduler()
+        e1 = rx.never()
+        e2 = rx.never()
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == []
+
+    def test_fork_join_never_empty(self):
+        scheduler = TestScheduler()
+        e1 = rx.never()
+        e2 = rx.empty()
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_completed(200)]
+
+    def test_fork_join_never_non_empty(self):
+        scheduler = TestScheduler()
+        e1 = rx.never()
+        e2 = scheduler.create_hot_observable([
+            on_next(150, 1),
+            on_next(230, 2),
+            on_completed(300)
+        ])
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == []
+
+    def test_fork_join_empty_empty(self):
+        scheduler = TestScheduler()
+        e1 = rx.empty()
+        e2 = rx.empty()
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_completed(200)]
+
+    def test_fork_join_empty_non_empty(self):
+        scheduler = TestScheduler()
+        e1 = rx.empty()
+        e2 = scheduler.create_hot_observable([
+            on_next(150, 1),
+            on_next(230, 2),
+            on_completed(300)
+        ])
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_completed(200)]
+
+    def test_fork_join_non_empty_non_empty_right_last(self):
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(215, 2), on_completed(230)]
+        msgs2 = [on_next(150, 1), on_next(220, 3), on_completed(240)]
+
+        e1 = scheduler.create_hot_observable(msgs1)
+        e2 = scheduler.create_hot_observable(msgs2)
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_next(240, (2, 3)), on_completed(240)]
+
+    def test_fork_join_non_empty_non_empty_left_last(self):
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(215, 2), on_completed(300)]
+        msgs2 = [on_next(150, 1), on_next(220, 3), on_completed(240)]
+
+        e1 = scheduler.create_hot_observable(msgs1)
+        e2 = scheduler.create_hot_observable(msgs2)
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_next(300, (2, 3)), on_completed(300)]
+
+    def test_fork_join_empty_error(self):
+        ex = RxException()
+
+        scheduler = TestScheduler()
+        e1 = rx.empty()
+        e2 = scheduler.create_hot_observable([
+            on_next(150, 1),
+            on_next(230, 2),
+            on_error(300, ex)
+        ])
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_completed(200)]
+
+    def test_fork_join_never_error(self):
+        ex = RxException()
+
+        scheduler = TestScheduler()
+        e1 = rx.never()
+        e2 = scheduler.create_hot_observable([
+            on_next(150, 1),
+            on_next(230, 2),
+            on_error(300, ex)
+        ])
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_error(300, ex)]
+
+    def test_fork_join_non_empty_error_left_last(self):
+        ex = RxException()
+
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(250, 2), on_completed(330)]
+        msgs2 = [on_next(150, 1), on_next(230, 2), on_error(300, ex)]
+        e1 = scheduler.create_hot_observable(msgs1)
+        e2 = scheduler.create_hot_observable(msgs2)
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_error(300, ex)]
+
+    def test_fork_join_non_empty_error_right_last(self):
+        ex = RxException()
+
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(250, 2), on_completed(300)]
+        msgs2 = [on_next(150, 1), on_next(230, 2), on_error(330, ex)]
+        e1 = scheduler.create_hot_observable(msgs1)
+        e2 = scheduler.create_hot_observable(msgs2)
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_error(330, ex)]
+
+    def test_fork_join_error_error_left_last(self):
+        ex = RxException()
+
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(250, 2), on_error(340, ex)]
+        msgs2 = [on_next(150, 1), on_next(230, 2), on_error(330, ex)]
+        e1 = scheduler.create_hot_observable(msgs1)
+        e2 = scheduler.create_hot_observable(msgs2)
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_error(330, ex)]
+
+    def test_fork_join_error_error_right_last(self):
+        ex = RxException()
+
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(250, 2), on_error(340, ex)]
+        msgs2 = [on_next(150, 1), on_next(230, 2), on_error(370, ex)]
+        e1 = scheduler.create_hot_observable(msgs1)
+        e2 = scheduler.create_hot_observable(msgs2)
+
+        results = scheduler.start(lambda: rx.fork_join(e1, e2))
+        assert results.messages == [on_error(340, ex)]
+
+    def test_fork_join_many(self):
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(210, 2), on_next(230, 3), on_next(300, 9), on_completed(500)]
+        msgs2 = [on_next(150, 1), on_next(205, 3), on_next(220, 7), on_next(400, 3), on_completed(900)]
+        msgs3 = [on_next(150, 1), on_next(250, 2), on_next(300, 3), on_next(400, 9), on_next(500, 2), on_completed(850)]
+        msgs4 = [on_next(150, 1), on_next(400, 2), on_next(550, 10), on_next(560, 11), on_next(600, 3),
+                 on_completed(605)]
+        msgs5 = [on_next(150, 1), on_next(201, 3), on_next(550, 10), on_next(560, 11), on_next(600, 3),
+                 on_next(900, 99), on_completed(905)]
+
+        xs = [scheduler.create_hot_observable(x) for x in [msgs1, msgs2, msgs3, msgs4, msgs5]]
+        results = scheduler.start(lambda: rx.fork_join(*xs))
+        assert results.messages == [on_next(905, (9, 3, 2, 3, 99)), on_completed(905)]
+
+    def test_fork_join_many_ops(self):
+        scheduler = TestScheduler()
+
+        msgs1 = [on_next(150, 1), on_next(210, 2), on_next(230, 3), on_next(300, 9), on_completed(500)]
+        msgs2 = [on_next(150, 1), on_next(205, 3), on_next(220, 7), on_next(400, 3), on_completed(900)]
+        msgs3 = [on_next(150, 1), on_next(250, 2), on_next(300, 3), on_next(400, 9), on_next(500, 2), on_completed(850)]
+        msgs4 = [on_next(150, 1), on_next(400, 2), on_next(550, 10), on_next(560, 11), on_next(600, 3),
+                 on_completed(605)]
+        msgs5 = [on_next(150, 1), on_next(201, 3), on_next(550, 10), on_next(560, 11), on_next(600, 3),
+                 on_next(900, 99), on_completed(905)]
+
+        xs = [scheduler.create_hot_observable(x) for x in [msgs2, msgs3, msgs4, msgs5]]
+
+        def create():
+            return scheduler.create_hot_observable(msgs1).pipe(ops.fork_join(*xs))
+
+        results = scheduler.start(create)
+        assert results.messages == [on_next(905, (9, 3, 2, 3, 99)), on_completed(905)]
diff --git a/tests/test_observable/test_tofuture.py b/tests/test_observable/test_tofuture.py
index 679c92a2..f90fb111 100644
--- a/tests/test_observable/test_tofuture.py
+++ b/tests/test_observable/test_tofuture.py
@@ -6,6 +6,7 @@ import rx
 import rx.operators as ops
 from rx.internal.exceptions import SequenceContainsNoElementsError
 from rx.testing import ReactiveTest
+from rx.subject import Subject
 
 on_next = ReactiveTest.on_next
 on_completed = ReactiveTest.on_completed
@@ -79,3 +80,33 @@ class TestToFuture(unittest.TestCase):
 
         loop.run_until_complete(go())
         assert result == 42
+
+    def test_cancel(self):
+        loop = asyncio.get_event_loop()
+
+        async def go():
+            source = rx.return_value(42)
+            fut = next(source.__await__())
+            # This used to raise an InvalidStateError before we got
+            # support for cancellation.
+            fut.cancel()
+            await fut
+
+        self.assertRaises(asyncio.CancelledError, loop.run_until_complete, go())
+
+    def test_dispose_on_cancel(self):
+        loop = asyncio.get_event_loop()
+        sub = Subject()
+
+        async def using_sub():
+            # Since the subject never completes, this await statement
+            # will never be complete either. We wait forever.
+            await rx.using(lambda: sub, lambda s: s)
+
+        async def go():
+            await asyncio.wait_for(using_sub(), 0.1)
+
+        self.assertRaises(asyncio.TimeoutError, loop.run_until_complete, go())
+        # When we cancel the future (due to the time-out), the future
+        # automatically disposes the underlying subject.
+        self.assertTrue(sub.is_disposed)
diff --git a/tests/test_observable/test_zip.py b/tests/test_observable/test_zip.py
index 41eda422..fe09b461 100644
--- a/tests/test_observable/test_zip.py
+++ b/tests/test_observable/test_zip.py
@@ -13,7 +13,6 @@ disposed = ReactiveTest.disposed
 created = ReactiveTest.created
 
 
-
 class TestZip(unittest.TestCase):
 
     def test_zip_never_never(self):
@@ -37,7 +36,7 @@ class TestZip(unittest.TestCase):
             return o1.pipe(ops.zip(o2))
 
         results = scheduler.start(create)
-        assert results.messages == []
+        assert results.messages == [on_completed(210)]
 
     def test_zip_empty_empty(self):
         scheduler = TestScheduler()
@@ -67,7 +66,7 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create)
-        assert results.messages == [on_completed(215)]
+        assert results.messages == [on_completed(210)]
 
     def test_zip_non_empty_empty(self):
         scheduler = TestScheduler()
@@ -82,7 +81,7 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create)
-        assert results.messages == [on_completed(215)]
+        assert results.messages == [on_completed(210)]
 
     def test_zip_never_non_empty(self):
         scheduler = TestScheduler()
@@ -96,7 +95,7 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create)
-        assert results.messages == []
+        assert results.messages == [on_completed(220)]
 
     def test_zip_non_empty_never(self):
         scheduler = TestScheduler()
@@ -110,7 +109,7 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create)
-        assert results.messages == []
+        assert results.messages == [on_completed(220)]
 
     def test_zip_non_empty_non_empty(self):
         scheduler = TestScheduler()
@@ -125,7 +124,7 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create)
-        assert results.messages == [on_next(220, 2 + 3), on_completed(240)]
+        assert results.messages == [on_next(220, 2 + 3), on_completed(230)]
 
     def test_zip_empty_error(self):
         ex = 'ex'
@@ -246,6 +245,7 @@ class TestZip(unittest.TestCase):
             for i in range(5):
                 results.append(on_next(205 + i * 5, i))
             return results
+
         msgs1 = msgs1_factory()
 
         def msgs2_factory():
@@ -253,6 +253,7 @@ class TestZip(unittest.TestCase):
             for i in range(10):
                 results.append(on_next(205 + i * 8, i))
             return results
+
         msgs2 = msgs2_factory()
 
         length = min(len(msgs1), len(msgs2))
@@ -265,13 +266,13 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create).messages
-        assert(length == len(results))
+        assert (length == len(results))
         for i in range(length):
             _sum = msgs1[i].value.value + msgs2[i].value.value
             time = max(msgs1[i].time, msgs2[i].time)
-            assert(results[i].value.kind == 'N'
-                   and results[i].time == time
-                   and results[i].value.value == _sum)
+            assert (results[i].value.kind == 'N'
+                    and results[i].time == time
+                    and results[i].value.value == _sum)
 
     def test_zip_some_data_asymmetric2(self):
         scheduler = TestScheduler()
@@ -282,6 +283,7 @@ class TestZip(unittest.TestCase):
                 results.append(on_next(205 + i * 5, i))
 
             return results
+
         msgs1 = msgs1_factory()
 
         def msgs2_factory():
@@ -289,6 +291,7 @@ class TestZip(unittest.TestCase):
             for i in range(5):
                 results.append(on_next(205 + i * 8, i))
             return results
+
         msgs2 = msgs2_factory()
 
         length = min(len(msgs1), len(msgs2))
@@ -301,13 +304,13 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create).messages
-        assert(length == len(results))
+        assert (length == len(results))
         for i in range(length):
             _sum = msgs1[i].value.value + msgs2[i].value.value
             time = max(msgs1[i].time, msgs2[i].time)
-            assert(results[i].value.kind == 'N'
-                   and results[i].time == time
-                   and results[i].value.value == _sum)
+            assert (results[i].value.kind == 'N'
+                    and results[i].time == time
+                    and results[i].value.value == _sum)
 
     def test_zip_some_data_symmetric(self):
         scheduler = TestScheduler()
@@ -317,6 +320,7 @@ class TestZip(unittest.TestCase):
             for i in range(10):
                 results.append(on_next(205 + i * 5, i))
             return results
+
         msgs1 = msgs1_factory()
 
         def msgs2_factory():
@@ -324,6 +328,7 @@ class TestZip(unittest.TestCase):
             for i in range(10):
                 results.append(on_next(205 + i * 8, i))
             return results
+
         msgs2 = msgs2_factory()
 
         length = min(len(msgs1), len(msgs2))
@@ -336,13 +341,13 @@ class TestZip(unittest.TestCase):
                 ops.map(sum))
 
         results = scheduler.start(create).messages
-        assert(length == len(results))
+        assert (length == len(results))
         for i in range(length):
             _sum = msgs1[i].value.value + msgs2[i].value.value
             time = max(msgs1[i].time, msgs2[i].time)
-            assert(results[i].value.kind == 'N'
-                   and results[i].time == time
-                   and results[i].value.value == _sum)
+            assert (results[i].value.kind == 'N'
+                    and results[i].time == time
+                    and results[i].value.value == _sum)
 
     def test_zip_with_iterable_never_empty(self):
         scheduler = TestScheduler()
@@ -423,7 +428,7 @@ class TestZip(unittest.TestCase):
     def test_zip_with_iterable_non_empty_non_empty(self):
         scheduler = TestScheduler()
         n1 = scheduler.create_hot_observable(
-                on_next(150, 1), on_next(215, 2), on_completed(230))
+            on_next(150, 1), on_next(215, 2), on_completed(230))
         n2 = [3]
 
         def create():