Provide mechanism for client session loss to trigger a teardown

confluent_server/confluent/consoleserver.py

@@ -537,6 +537,8 @@ class ConsoleHandler(object):
         for rcpt in list(self.livesessions):
             try:
                 await rcpt.data_handler(data)
+            except exc.Disconnect:
+                await rcpt.destroy()
             except Exception as e:  # No matter the reason, advance to next recipient
                 print(repr(e))
                 _tracelog.log(traceback.format_exc(), ltype=log.DataTypes.event,

confluent_server/confluent/exceptions.py

@@ -146,3 +146,7 @@ class LoggedOut(ConfluentException):
 
     def get_error_body(self):
         return '{"loggedout": 1}'
+    
+class Disconnect(ConfluentException):
+    apierrorcode = 503
+    _apierrorstr = 'Client disconnected'

confluent_server/confluent/httpapi.py

@@ -17,6 +17,8 @@
 # This SCGI server provides a http wrap to confluent api
 # It additionally manages httprequest console sessions
 import base64
+
+import aiohttp
 try:
     import Cookie
 except ModuleNotFoundError:
@@ -449,7 +451,10 @@ def websockify_data(data):
 def datacallback_bound(clientsessid, rsp):
     async def datacallback(data):
         data = websockify_data(data)
-        await rsp.send_str(u'${0}$'.format(clientsessid) + data)
+        try:
+            await rsp.send_str(u'${0}$'.format(clientsessid) + data)
+        except aiohttp.client_exceptions.ClientConnectionResetError:
+            raise exc.Disconnect("Client disconnected")
     return datacallback
 
 async def wsock_handler(req):