Repository: ReactiveX/RxPY

Issue:
Async function with flat_map()
Observer does not recieve values if there is `flat_map()` operator in chain and mapper returns `Future`.

Rx: v3.0.1-2-g19b19087
Python: 3.7.4

Test code:
```python
from asyncio import get_event_loop, create_task, sleep

from rx import operators as op
from rx.core import Observer
from rx.subject import Subject
from rx.scheduler.eventloop import AsyncIOScheduler


def on_next(value, *args, **kwargs):
    print('Recieve', value)


def maptest(value):
    async def increase(value):
        print('Increasing')
        return value + 1

    print('Returning task')
    return create_task(increase(value))


async def go(loop):
    print('Starting, creating stream')
    stream = Subject()
    tail = stream.pipe(op.flat_map(maptest))
    tail.subscribe(Observer(on_next), scheduler=AsyncIOScheduler(loop))

    print('Sending value')
    stream.on_next(10)

    print('Sleeping 5 sec')
    await sleep(5)
    print('End of sleeping')

loop = get_event_loop()
loop.run_until_complete(go(loop))
print('Happy end')
```
Result output:
```
Starting, creating stream
Sending value
Returning task
Sleeping 5 sec
Increasing
End of sleeping
Happy end
```

This is because of type checking in `_flat_map_internal()` from [flat_map()](../blob/cd3d5a5b833a0df3ecd91116c0243ebaa762578c/rx/core/operators/flatmap.py#L13-L18) operator:
```python
        if isinstance(mapper_result, collections.abc.Iterable):
            result = from_(mapper_result)
        else:
            result = from_future(mapper_result) if is_future(
                mapper_result) else mapper_result
        return result
```

If mapper returns `Future` Observable will create with `from_` operator, not with `from_future` operator, because:

```python
>>> issubclass(asyncio.Future, collections.abc.Iterable)
True
>>> issubclass(asyncio.Task, asyncio.Future)
True
```

So we need rewrite `mapper_result` type checking, for example:
```python
        if is_future(mapper_result):
            result = from_future(mapper_result)
        elif isinstance(mapper_result, collections.abc.Iterable):
            result = from_(mapper_result)
        else:
            result = mapper_result
        return result
```

Now all is ok, test output:
```
Starting, creating stream
Sending value
Returning task
Sleeping 5 sec
Increasing
Recieve 11
End of sleeping
Happy end
```

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

Coverage increased (+0.002%) to 92.904% when pulling **d7c7f00982a88193214883a4036d80099074cb32 on MainRo:bugfix/flatmap-future-mapper** into **19b1908724b5a4b7ef3e6c407d52cdc3daaa4ed0 on ReactiveX:master**.


Patch:
diff --git a/rx/core/operators/flatmap.py b/rx/core/operators/flatmap.py
index 0f6b73d8..f0262658 100644
--- a/rx/core/operators/flatmap.py
+++ b/rx/core/operators/flatmap.py
@@ -10,11 +10,12 @@ from rx.internal.utils import is_future
 def _flat_map_internal(source, mapper=None, mapper_indexed=None):
     def projection(x, i):
         mapper_result = mapper(x) if mapper else mapper_indexed(x, i)
-        if isinstance(mapper_result, collections.abc.Iterable):
+        if is_future(mapper_result):
+            result = from_future(mapper_result)
+        elif isinstance(mapper_result, collections.abc.Iterable):
             result = from_(mapper_result)
         else:
-            result = from_future(mapper_result) if is_future(
-                mapper_result) else mapper_result
+            result = mapper_result
         return result
 
     return source.pipe(


Test patch:
diff --git a/tests/test_observable/test_flatmap_async.py b/tests/test_observable/test_flatmap_async.py
new file mode 100644
index 00000000..f568638e
--- /dev/null
+++ b/tests/test_observable/test_flatmap_async.py
@@ -0,0 +1,33 @@
+import unittest
+import asyncio
+from rx import operators as ops
+from rx.subject import Subject
+
+from rx.scheduler.eventloop import AsyncIOScheduler
+
+
+class TestFlatMapAsync(unittest.TestCase):
+
+    def test_flat_map_async(self):
+        actual_next = None
+        loop = asyncio.get_event_loop()
+        scheduler = AsyncIOScheduler(loop=loop)
+
+        def mapper(i):
+            async def _mapper(i):
+                return i + 1
+
+            return asyncio.ensure_future(_mapper(i))
+
+        def on_next(i):
+            nonlocal actual_next
+            actual_next = i
+
+        async def test_flat_map():
+            x = Subject()
+            x.pipe(ops.flat_map(mapper)).subscribe(on_next, scheduler=scheduler)
+            x.on_next(10)
+            await asyncio.sleep(0.1)
+
+        loop.run_until_complete(test_flat_map())
+        assert actual_next == 11