fire and forget, close stream only on future done.

Author: jell-o-fishi

CHANGELOG.rst

@@ -1,6 +1,10 @@
 Changelog
 ---------
 
+v0.4.6
+======
+- fire_and_forget now only removes the stream id when the future is done.
+
 v0.4.5
 ======
 - Breaking change: Normalized the request_channel method argument names across implementations and added where missing (vanilla, reactivex etc.):

rsocket/rsocket_base.py

@@ -493,7 +493,7 @@ def fire_and_forget(self, payload: Payload) -> Awaitable[None]:
         stream_id = self._allocate_stream()
         frame = to_fire_and_forget_frame(stream_id, payload)
         self.send_request(frame)
-        self.finish_stream(stream_id)
+        frame.sent_future.add_done_callback(lambda _: self.finish_stream(stream_id))
         return frame.sent_future
 
     def request_stream(self, payload: Payload) -> Union[BackpressureApi, Publisher]:

tests/rsocket/test_misbehaving_client.py

@@ -29,9 +29,9 @@ async def request_response(self, payload: Payload) -> Awaitable[Payload]:
 
     bad_client = MisbehavingRSocket(client._transport)
 
-    client.fire_and_forget(Payload())
+    await client.fire_and_forget(Payload())
 
-    await bad_client.send_frame(to_payload_frame(145, Payload()))
+    await bad_client.send_frame(to_payload_frame(stream_id=145, payload=Payload()))
 
     await client.request_response(Payload(b'request'))
 

tests/rsocket/test_routing.py

@@ -153,7 +153,7 @@ async def fire_and_forget(payload):
     async with lazy_pipe(
             client_arguments={'metadata_encoding': WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA},
             server_arguments={'handler_factory': handler_factory}) as (server, client):
-        client.fire_and_forget(Payload(b'request data', composite(route('test.path'))))
+        await client.fire_and_forget(Payload(b'request data', composite(route('test.path'))))
 
         await received.wait()
         assert received_data == b'request data'