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**.

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(

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
```
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

pip install -e .
parse_log_pytest
pytest
pytest pytest-asyncio
3.9
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
exceptiongroup @ file:///croot/exceptiongroup_1706031385326/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
packaging @ file:///croot/packaging_1734472117206/work
pluggy @ file:///croot/pluggy_1733169602837/work
pytest @ file:///croot/pytest_1738938843180/work
pytest-asyncio==0.26.0
-e git+https://github.com/ReactiveX/RxPY.git@573260422a6e3825c82a1cfbf8260b6514afc7b6#egg=Rx
tomli @ file:///opt/conda/conda-bld/tomli_1657175507142/work
typing_extensions==4.13.0

name: RxPY
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - exceptiongroup=1.2.0=py39h06a4308_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - packaging=24.2=py39h06a4308_0
  - pip=25.0=py39h06a4308_0
  - pluggy=1.5.0=py39h06a4308_0
  - pytest=8.3.4=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tomli=2.0.1=py39h06a4308_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - pytest-asyncio==0.26.0
      - typing-extensions==4.13.0
prefix: /opt/conda/envs/RxPY

tests/test_observable/test_flatmap_async.py::TestFlatMapAsync::test_flat_map_async
rx/core/operators/flatmap.py
rx/core/operators/flatmap.py