Repository: ReactiveX/RxPY

Issue:
count() doesn't seem to work with group_by()
Hi all,
while trying this library I came across what seems to be a bug. I'm trying to group a stream of integers ranging from 0 to 9 by being odd or even.

```python
from rx import from_, range as rxrange
from rx.operators import group_by, count


def even_or_odd(x):
    if x % 2 == 0:
        return 'even'
    return 'odd'


def group_consumer(group_observable):
    def print_number(n):
        print(f"Observable {group_observable.key}: count {n}")

    group_observable.pipe(count()).subscribe(print_number)


print("---> With from_")
test1 = from_(range(10)).pipe(group_by(even_or_odd)).subscribe(group_consumer)
print("\n---> With rx.range")
test2 = rxrange(0, 10, 1).pipe(group_by(even_or_odd)).subscribe(group_consumer)

```

The output is:

```bash
---> With from_
Observable even: count 0
Observable odd: count 0

---> With rx.range
Observable even: count 3
Observable odd: count 3

Process finished with exit code 0
```

Two things are odd in the output:

1. The result is different depending upon the usage of `from_` or `rx.range` (they should do the same, right?)
2. In both cases the result is wrong (should be 5 and 5, right?)

I hope I am not making some silly mistake,
thank you for your help.

Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index e10f1846..2131a27f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -23,11 +23,11 @@ matrix:
         - xvfb-run -a python3 setup.py test
 
 
-    - name: "Python 3.8-dev on Linux"
+    - name: "Python 3.8 on Linux"
       os: linux
       dist: xenial
       language: python
-      python: 3.8-dev
+      python: 3.8
       service:
         - xvfb
       script:
diff --git a/rx/core/observable/fromfuture.py b/rx/core/observable/fromfuture.py
index fa83acbc..70ab7670 100644
--- a/rx/core/observable/fromfuture.py
+++ b/rx/core/observable/fromfuture.py
@@ -1,3 +1,4 @@
+import asyncio
 from asyncio.futures import Future
 from typing import Optional
 
@@ -26,7 +27,7 @@ def _from_future(future: Future) -> Observable:
         def done(future):
             try:
                 value = future.result()
-            except Exception as ex:  # pylint: disable=broad-except
+            except (Exception, asyncio.CancelledError) as ex:  # pylint: disable=broad-except
                 observer.on_error(ex)
             else:
                 observer.on_next(value)
diff --git a/rx/core/operators/reduce.py b/rx/core/operators/reduce.py
index 0fbc4196..112514e6 100644
--- a/rx/core/operators/reduce.py
+++ b/rx/core/operators/reduce.py
@@ -5,6 +5,7 @@ from rx.internal.utils import NotSet
 from rx.core import Observable, pipe
 from rx.core.typing import Accumulator
 
+
 def _reduce(accumulator: Accumulator, seed: Any = NotSet) -> Callable[[Observable], Observable]:
     """Applies an accumulator function over an observable sequence,
     returning the result of the aggregation as a single element in the
@@ -28,12 +29,8 @@ def _reduce(accumulator: Accumulator, seed: Any = NotSet) -> Callable[[Observabl
         an observable sequence containing a single element with the
         final accumulator value.
     """
-
-
     if seed is not NotSet:
-        initial = ops.start_with(seed)
         scanner = ops.scan(accumulator, seed=seed)
-
-        return pipe(scanner, initial, ops.last())
+        return pipe(scanner, ops.last_or_default(default_value=seed))
 
     return pipe(ops.scan(accumulator), ops.last())
diff --git a/rx/core/operators/todict.py b/rx/core/operators/todict.py
index 6489af38..17d456a7 100644
--- a/rx/core/operators/todict.py
+++ b/rx/core/operators/todict.py
@@ -39,7 +39,9 @@ def _to_dict(key_mapper: Mapper,
                 m[key] = element
 
             def on_completed() -> None:
+                nonlocal m
                 observer.on_next(m)
+                m = dict()
                 observer.on_completed()
 
             return source.subscribe_(on_next, observer.on_error, on_completed, scheduler)
diff --git a/rx/core/operators/tofuture.py b/rx/core/operators/tofuture.py
index d9cad7a7..67197888 100644
--- a/rx/core/operators/tofuture.py
+++ b/rx/core/operators/tofuture.py
@@ -41,10 +41,12 @@ def _to_future(future_ctor: Optional[Callable[[], Future]] = None,
             future.set_exception(err)
 
         def on_completed():
+            nonlocal last_value
             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)
 
diff --git a/rx/core/operators/toiterable.py b/rx/core/operators/toiterable.py
index ec8cd6f5..a7b811eb 100644
--- a/rx/core/operators/toiterable.py
+++ b/rx/core/operators/toiterable.py
@@ -20,7 +20,9 @@ def _to_iterable() -> Callable[[Observable], Observable]:
                 queue.append(item)
 
             def on_completed():
+                nonlocal queue
                 observer.on_next(queue)
+                queue = []
                 observer.on_completed()
 
             return source.subscribe_(on_next, observer.on_error, on_completed, scheduler)
diff --git a/rx/core/operators/toset.py b/rx/core/operators/toset.py
index d34fc1c2..d9bbcec0 100644
--- a/rx/core/operators/toset.py
+++ b/rx/core/operators/toset.py
@@ -14,7 +14,9 @@ def _to_set() -> Callable[[Observable], Observable]:
             s = set()
 
             def on_completed():
+                nonlocal s
                 observer.on_next(s)
+                s = set()
                 observer.on_completed()
 
             return source.subscribe_(s.add, observer.on_error, on_completed, scheduler)
diff --git a/rx/operators/__init__.py b/rx/operators/__init__.py
index c4d5afbe..c2373d76 100644
--- a/rx/operators/__init__.py
+++ b/rx/operators/__init__.py
@@ -870,7 +870,7 @@ def expand(mapper: Mapper) -> Callable[[Observable], Observable]:
 
 def filter(predicate: Predicate) -> Callable[[Observable], Observable]:
     """Filters the elements of an observable sequence based on a
-    predicate by incorporating the element's index.
+    predicate.
 
     .. marble::
         :alt: filter
@@ -2019,7 +2019,7 @@ def repeat(repeat_count: Optional[int] = None) -> Callable[[Observable], Observa
 
     Returns:
         An operator function that takes an observable sources and
-        returna an observable sequence producing the elements of the
+        returns an observable sequence producing the elements of the
         given sequence repeatedly.
     """
     from rx.core.operators.repeat import _repeat


Test patch:
diff --git a/tests/test_integration/test_group_reduce.py b/tests/test_integration/test_group_reduce.py
new file mode 100644
index 00000000..87d8d135
--- /dev/null
+++ b/tests/test_integration/test_group_reduce.py
@@ -0,0 +1,30 @@
+import unittest
+
+import rx
+from rx import operators as ops
+
+
+class TestGroupByReduce(unittest.TestCase):
+    def test_groupby_count(self):
+        res = []
+        counts = rx.from_(range(10)).pipe(
+            ops.group_by(lambda i: 'even' if i % 2 == 0 else 'odd'),
+            ops.flat_map(lambda i: i.pipe(
+                ops.count(),
+                ops.map(lambda ii: (i.key, ii)),
+            ))
+        )
+
+        counts.subscribe(on_next=res.append)
+        assert res == [('even', 5), ('odd', 5)]
+
+    def test_window_sum(self):
+        res = []
+        rx.from_(range(6)).pipe(
+            ops.window_with_count(count=3, skip=1),
+            ops.flat_map(lambda i: i.pipe(
+                ops.sum(),
+            )),
+        ).subscribe(on_next=res.append)
+
+        assert res == [3, 6, 9, 12, 9, 5, 0]
diff --git a/tests/test_observable/test_fromfuture.py b/tests/test_observable/test_fromfuture.py
index 17493fcf..f4ac2eb7 100644
--- a/tests/test_observable/test_fromfuture.py
+++ b/tests/test_observable/test_fromfuture.py
@@ -59,6 +59,30 @@ class TestFromFuture(unittest.TestCase):
         loop.run_until_complete(go())
         assert all(success)
 
+    def test_future_cancel(self):
+        loop = asyncio.get_event_loop()
+        success = [True, False, True]
+
+        @asyncio.coroutine
+        def go():
+            future = Future()
+            source = rx.from_future(future)
+
+            def on_next(x):
+                success[0] = False
+
+            def on_error(err):
+                success[1] = type(err) == asyncio.CancelledError
+
+            def on_completed():
+                success[2] = False
+
+            source.subscribe(on_next, on_error, on_completed)
+            future.cancel()
+
+        loop.run_until_complete(go())
+        assert all(success)
+
     def test_future_dispose(self):
         loop = asyncio.get_event_loop()
         success = [True, True, True]