@@ -142,13 +142,21 @@ async def _inner_send(
142142 callback = self ._connection ._send_message_to_server (
143143 self ._object , method , augmented_params , timeout
144144 )
145- done , _ = await asyncio .wait (
146- {
147- self ._connection ._transport .on_error_future ,
148- callback .future ,
149- },
150- return_when = asyncio .FIRST_COMPLETED ,
151- )
145+ try :
146+ done , _ = await asyncio .wait (
147+ {
148+ self ._connection ._transport .on_error_future ,
149+ callback .future ,
150+ },
151+ return_when = asyncio .FIRST_COMPLETED ,
152+ )
153+ except asyncio .CancelledError as exc :
154+ await self ._connection ._abort (
155+ self ._object ,
156+ callback ,
157+ str (exc ) or "Task was cancelled" ,
158+ )
159+ raise
152160 if not callback .future .done ():
153161 callback .future .cancel ()
154162 result = next (iter (done )).result ()
@@ -240,7 +248,10 @@ def remove_listener(self, event: str, f: Any) -> None:
240248
241249
242250class ProtocolCallback :
243- def __init__ (self , loop : asyncio .AbstractEventLoop , no_reply : bool = False ) -> None :
251+ def __init__ (
252+ self , loop : asyncio .AbstractEventLoop , id : int , no_reply : bool = False
253+ ) -> None :
254+ self .id = id
244255 self .stack_trace : traceback .StackSummary
245256 self .no_reply = no_reply
246257 self .future = loop .create_future ()
@@ -389,7 +400,7 @@ def _send_message_to_server(
389400 )
390401 self ._last_id += 1
391402 id = self ._last_id
392- callback = ProtocolCallback (self ._loop , no_reply = no_reply )
403+ callback = ProtocolCallback (self ._loop , id , no_reply = no_reply )
393404 task = asyncio .current_task (self ._loop )
394405 callback .stack_trace = cast (
395406 traceback .StackSummary ,
@@ -433,6 +444,34 @@ def _send_message_to_server(
433444
434445 return callback
435446
447+ async def _abort (
448+ self , object : ChannelOwner , callback : ProtocolCallback , reason : str
449+ ) -> None :
450+ try :
451+ self ._transport .send (
452+ {
453+ "guid" : object ._guid ,
454+ "method" : "__abort__" ,
455+ "params" : {"id" : callback .id , "reason" : reason },
456+ }
457+ )
458+ except (Error , OSError ):
459+ pass
460+ try :
461+ done , _ = await asyncio .wait (
462+ {
463+ self ._transport .on_error_future ,
464+ callback .future ,
465+ },
466+ return_when = asyncio .FIRST_COMPLETED ,
467+ )
468+ finally :
469+ if not callback .future .done ():
470+ callback .future .cancel ()
471+ for future in done :
472+ if not future .cancelled ():
473+ future .exception ()
474+
436475 def dispatch (self , msg : ParsedMessagePayload ) -> None :
437476 if self ._closed_error :
438477 return
0 commit comments