1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
import asyncio
import unittest
from reactivex import operators as ops
from reactivex.scheduler.eventloop import AsyncIOScheduler
from reactivex.subject import Subject
class TestFlatMapAsync(unittest.TestCase):
def test_flat_map_async(self):
actual_next = None
loop = asyncio.new_event_loop()
scheduler = AsyncIOScheduler(loop=loop)
def mapper(i: int):
async def _mapper(i: int):
return i + 1
return asyncio.ensure_future(_mapper(i))
def on_next(i: int):
nonlocal actual_next
actual_next = i
def on_error(ex):
print("Error", ex)
async def test_flat_map():
x: Subject[int] = Subject()
x.pipe(ops.flat_map(mapper)).subscribe(
on_next, on_error, scheduler=scheduler
)
x.on_next(10)
await asyncio.sleep(0.1)
loop.run_until_complete(test_flat_map())
assert actual_next == 11
|
