Merge pull request #15 from SomberNight/finalrpcerror

FinalRPCError(RPCError): send error to client and then disconnect

aiorpcx/jsonrpc.py

@@ -27,7 +27,7 @@
 
 __all__ = ('JSONRPC', 'JSONRPCv1', 'JSONRPCv2', 'JSONRPCLoose',
            'JSONRPCAutoDetect', 'Request', 'Notification', 'Batch',
-           'RPCError', 'ProtocolError',
+           'RPCError', 'FinalRPCError', 'ProtocolError',
            'JSONRPCConnection', 'handler_invocation')
 
 import itertools
@@ -140,6 +140,10 @@ class RPCError(CodeMessageError):
     pass
 
 
+class FinalRPCError(RPCError):
+    pass
+
+
 class ProtocolError(CodeMessageError):
 
     def __init__(self, code, message):

aiorpcx/session.py

@@ -41,7 +41,7 @@
     BadMagicError, BadChecksumError, OversizedPayloadError
 )
 from aiorpcx.jsonrpc import (
-    Request, Batch, Notification, ProtocolError, RPCError,
+    Request, Batch, Notification, ProtocolError, RPCError, FinalRPCError,
     JSONRPC, JSONRPCv2, JSONRPCConnection
 )
 from aiorpcx.util import Concurrency
@@ -496,7 +496,10 @@ async def _throttled_request(self, request):
                 message = request.send_result(result)
                 if message:
                     await self._send_message(message)
-            if isinstance(result, Exception):
+            if isinstance(result, FinalRPCError):
+                # Don't await self.close() because that is self-cancelling
+                self._close()
+            elif isinstance(result, Exception):
                 self._bump_errors()
 
     def connection_lost(self, exc):

tests/test_session.py

@@ -63,6 +63,9 @@ async def on_notify(self, thing):
     async def on_bug(self):
         raise ValueError
 
+    async def on_incompatibleversion(self):
+        raise FinalRPCError(1, "incompatible version")
+
     async def on_sleepy(self):
         await sleep(10)
 
@@ -366,6 +369,14 @@ async def test_close_on_many_errors(self, server):
             pass
         assert server_session.errors == server_session.max_errors
 
+    @pytest.mark.asyncio
+    async def test_finalrpcerror(self, server):
+        async with Connector(RPCSession, 'localhost',
+                             server.port) as client:
+            with pytest.raises(RPCError) as e:
+                await client.send_request('incompatibleversion')
+            assert client.is_closing()
+
     @pytest.mark.asyncio
     async def test_send_empty_batch(self, server):
         async with Connector(RPCSession, 'localhost', server.port) as client: