handling situations when web socket server or client are disconnected

Author: vesellov

bitdust/interface/routed_web_socket.py

@@ -167,10 +167,12 @@ def stop_client(url):
         except Empty:
             break
     _WebSocketQueue[url].put_nowait((None, None))
-    if ws(url):
+    _ws = ws(url)
+    if _ws:
+        _ws.close()
+    else:
         if _Debug:
             lg.dbg(_DebugLevel, 'websocket %s already closed' % url)
-        ws(url).close()
 
 
 def shutdown_clients():
@@ -423,6 +425,7 @@ def init(self):
         self.handshaked_routers = []
         self.server_code = None
         self.device_name = None
+        self.client_connected = False
 
     def state_changed(self, oldstate, newstate, event, *args, **kwargs):
         """
@@ -442,6 +445,7 @@ def to_json(self, short=True):
             'connected_routers': self.handshaked_routers,
             'server_code': self.server_code,
             'device_name': self.device_name,
+            'client_connected': self.client_connected,
         })
         return ret
 
@@ -456,6 +460,7 @@ def on_incoming_message(self, url, json_data):
         cmd = json_data.get('cmd')
         if cmd == 'api':
             self.active_router_url = url
+            self.client_connected = True
             self.event('api-message', url=url, json_data=json_data)
             return True
         if cmd == 'handshake-accepted':
@@ -485,11 +490,17 @@ def on_incoming_message(self, url, json_data):
                 return False
             self.on_server_code_received(signature=signature, encrypted_server_code=encrypted_server_code)
             return True
+        if cmd == 'client-disconnected':
+            self.client_connected = False
+            return True
         return False
 
     def on_outgoing_message(self, json_data):
         if _Debug:
-            lg.args(_DebugLevel, json_data)
+            lg.args(_DebugLevel, client_connected=self.client_connected, d=json_data)
+        if json_data.get('cmd') == 'push':
+            if not self.client_connected:
+                return False
         if self.state != 'READY':
             lg.warn('skip sending api message to client, %r state is %r' % (self, self.state))
             return False

bitdust/interface/web_socket_transmitter.py

@@ -345,10 +345,10 @@ def buildProtocol(self, addr):
 def do_process_incoming_message(transport, json_data, raw_data):
     global MAX_KNOWN_ROUTES
     route_id = transport.route_id
-    direction = transport.direction
+    transport_direction = transport.direction
     if _Debug:
-        lg.args(_DebugLevel, route_id=route_id, direction=direction, json_data=json_data)
-    if route_id is None and direction is None:
+        lg.args(_DebugLevel, route_id=route_id, direction=transport_direction, json_data=json_data)
+    if route_id is None and transport_direction is None:
         if json_data.get('cmd') == 'connect-request':
             if len(routes()) >= MAX_KNOWN_ROUTES:
                 if _Debug:
@@ -368,18 +368,18 @@ def do_process_incoming_message(transport, json_data, raw_data):
                 lg.exc()
                 return False
             if _Debug:
-                lg.out(_DebugLevel, '    wrote %d bytes to %s/%s at %r' % (len(response_raw_data), route_id, direction, transport))
+                lg.out(_DebugLevel, '    wrote %d bytes to %s/%s at %r' % (len(response_raw_data), route_id, transport_direction, transport))
             return True
         return False
     if not route_id:
         lg.warn('unknown route ID: %r' % transport)
         return False
-    if not direction:
+    if not transport_direction:
         lg.warn('route direction was not identified: %r' % transport)
         return False
     cmd = json_data.get('cmd')
     if cmd == 'handshake':
-        if json_data.get('internal') and direction == 'internal':
+        if json_data.get('internal') and transport_direction == 'internal':
             route_info = routes(route_id)
             route_url = route_info['route_url'] if route_info else None
             if not route_url:
@@ -396,7 +396,7 @@ def do_process_incoming_message(transport, json_data, raw_data):
                 lg.exc()
                 return False
             if _Debug:
-                lg.out(_DebugLevel, '    wrote %d bytes to %s/%s at %r' % (len(response_raw_data), route_id, direction, transport))
+                lg.out(_DebugLevel, '    wrote %d bytes to %s/%s at %r' % (len(response_raw_data), route_id, transport_direction, transport))
             return True
     call_id = json_data.get('call_id')
     if call_id or (cmd in (
@@ -410,15 +410,28 @@ def do_process_incoming_message(transport, json_data, raw_data):
     )):
         route_info = routes(route_id)
         if _Debug:
-            lg.args(_DebugLevel, direction=direction, route_info=route_info)
+            lg.args(_DebugLevel, direction=transport_direction, route_info=route_info)
         if not route_info:
             lg.warn('route info for %r was not found' % route_id)
             return False
-        if direction == 'external':
-            internal_transport = route_info.get('internal_transport')
+        external_transport = route_info.get('external_transport')
+        internal_transport = route_info.get('internal_transport')
+        if transport_direction == 'external':
             if not internal_transport:
+                if not external_transport:
+                    lg.warn('internal and external transport for route %r are both disconnected' % route_id)
+                    return False
                 lg.warn('internal transport for route %r is not connected' % route_id)
-                return False
+                json_response = {
+                    'cmd': 'server-disconnected',
+                }
+                response_raw_data = serialization.DictToBytes(json_response, encoding='utf-8')
+                try:
+                    transport.write(response_raw_data)
+                except:
+                    lg.exc()
+                    return False
+                return True
             try:
                 internal_transport.write(raw_data)
             except:
@@ -427,11 +440,22 @@ def do_process_incoming_message(transport, json_data, raw_data):
             if _Debug:
                 lg.out(_DebugLevel, '    wrote %d bytes to %s/%s at %r' % (len(raw_data), route_id, 'internal', internal_transport))
             return True
-        if direction == 'internal':
-            external_transport = route_info.get('external_transport')
+        if transport_direction == 'internal':
             if not external_transport:
+                if not internal_transport:
+                    lg.warn('external and internal transport for route %r are both disconnected' % route_id)
+                    return False
                 lg.warn('external transport for route %r is not connected' % route_id)
-                return False
+                json_response = {
+                    'cmd': 'client-disconnected',
+                }
+                response_raw_data = serialization.DictToBytes(json_response, encoding='utf-8')
+                try:
+                    transport.write(response_raw_data)
+                except:
+                    lg.exc()
+                    return False
+                return True
             try:
                 external_transport.write(raw_data)
             except:
@@ -440,5 +464,5 @@ def do_process_incoming_message(transport, json_data, raw_data):
             if _Debug:
                 lg.out(_DebugLevel, '    wrote %d bytes to %s/%s at %r' % (len(raw_data), route_id, 'external', external_transport))
             return True
-        lg.warn('unexpected direction: %r' % direction)
+        lg.warn('unexpected direction: %r' % transport_direction)
     return False