-
Notifications
You must be signed in to change notification settings - Fork 23
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #104 from rsocket/disposable
Disposable
- Loading branch information
Showing
1 changed file
with
63 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,63 @@ | ||
import asyncio | ||
from time import sleep | ||
from typing import Tuple | ||
|
||
import pytest | ||
import reactivex | ||
from reactivex import operators, Observable | ||
from reactivex.operators._tofuture import to_future_ | ||
from reactivex.scheduler import ThreadPoolScheduler | ||
from reactivex.scheduler.eventloop import AsyncIOScheduler | ||
|
||
from rsocket.exceptions import RSocketProtocolError | ||
from rsocket.frame_helpers import ensure_bytes | ||
from rsocket.payload import Payload | ||
from rsocket.reactivex.reactivex_client import ReactiveXClient | ||
from rsocket.reactivex.reactivex_handler import BaseReactivexHandler | ||
from rsocket.reactivex.reactivex_handler_adapter import reactivex_handler_factory | ||
from rsocket.rsocket_client import RSocketClient | ||
from rsocket.rsocket_server import RSocketServer | ||
|
||
|
||
async def test_serve_reactivex_stream_disconnected(pipe: Tuple[RSocketServer, RSocketClient]): | ||
sent_counter = 0 | ||
|
||
def increment_sent_counter(): | ||
nonlocal sent_counter | ||
sent_counter += 1 | ||
|
||
def delayed(message): | ||
sleep(0.3) | ||
return message | ||
|
||
class Handler(BaseReactivexHandler): | ||
|
||
async def request_stream(self, payload: Payload) -> Observable: | ||
return reactivex.from_((delayed('Feed Item: {}'.format(index)) for index in range(10)), | ||
ThreadPoolScheduler()).pipe( | ||
operators.do_action(on_next=lambda _: increment_sent_counter()), | ||
operators.map(lambda _: Payload(ensure_bytes(_))) | ||
) | ||
|
||
server, client = pipe | ||
|
||
server.set_handler_using_factory(reactivex_handler_factory(Handler)) | ||
|
||
async def request(): | ||
await ReactiveXClient(client).request_stream(Payload(b'request text'), | ||
request_limit=2).pipe( | ||
operators.map(lambda payload: payload.data), | ||
operators.to_list(), | ||
to_future_(scheduler=AsyncIOScheduler(loop=asyncio.get_event_loop())) | ||
) | ||
|
||
task = asyncio.create_task(request()) | ||
|
||
await asyncio.sleep(1) | ||
|
||
await client.close() | ||
|
||
assert 0 < sent_counter < 5 | ||
|
||
with pytest.raises(RSocketProtocolError): | ||
await task |