Fix half-open socket (#54)

* add blade.ping and test it

* use blade.ping to detect half-open socket connection

* minor refactoring and tests

* changelog entry

* increase keepalive delay

* use const for ping_delay

Author: Edoardo Gallo

CHANGELOG.md

@@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Fixed
 - Check signals supported by the environment. On Windows there is no `SIGHUP`.
+- Detect half-open connection and force close connection to update Client/Consumer properly.
 
 ## [2.0.0rc1] - 2019-10-28
 ### Added

signalwire/blade/messages/ping.py

@@ -0,0 +1,8 @@
+from signalwire.blade.messages.message import Message
+
+class Ping(Message):
+
+  def __init__(self, timestamp=None):
+    self.method = 'blade.ping'
+    params = { 'timestamp': timestamp } if timestamp else {}
+    super().__init__(params=params)

signalwire/relay/client.py

@@ -6,6 +6,7 @@
 from uuid import uuid4
 from signalwire.blade.connection import Connection
 from signalwire.blade.messages.connect import Connect
+from signalwire.blade.messages.ping import Ping
 from signalwire.blade.handler import register, unregister, trigger
 from .helpers import setup_protocol
 from .calling import Calling
@@ -15,6 +16,8 @@
 from .constants import Constants, WebSocketEvents
 
 class Client:
+  PING_DELAY = 10
+
   def __init__(self, project, token, host=Constants.HOST, connection=Connection):
     self.loop = asyncio.get_event_loop()
     self.host = host
@@ -34,6 +37,7 @@ def __init__(self, project, token, host=Constants.HOST, connection=Connection):
     self._requests = {}
     self._idle = False
     self._executeQueue = asyncio.Queue()
+    self._pingInterval = None
     log_level = os.getenv('LOG_LEVEL', 'INFO').upper()
     logging.basicConfig(level=log_level)
 
@@ -79,7 +83,7 @@ async def _connect(self):
       await self.connection.connect()
       asyncio.create_task(self.on_socket_open())
       await self.connection.read()
-      trigger(WebSocketEvents.CLOSE, suffix=self.uuid)
+      self.on_socket_close()
     except aiohttp.client_exceptions.ClientConnectorError as error:
       trigger(WebSocketEvents.ERROR, error, suffix=self.uuid)
       logging.warn(f"{self.host} seems down..")
@@ -122,6 +126,12 @@ def attach_signals(self):
       except:
         pass
 
+  def on_socket_close(self):
+    if self._pingInterval:
+      self._pingInterval.cancel()
+    self.contexts = []
+    trigger(WebSocketEvents.CLOSE, suffix=self.uuid)
+
   async def on_socket_open(self):
     try:
       self._idle = False
@@ -131,12 +141,24 @@ async def on_socket_open(self):
       self.signature = result['authorization']['signature']
       self.protocol = await setup_protocol(self)
       await self._clearExecuteQueue()
+      self._pong = True
+      self.keepalive()
       logging.info('Client connected!')
       trigger(Constants.READY, self, suffix=self.uuid)
     except Exception as error:
       logging.error('Client setup error: {0}'.format(str(error)))
       await self.connection.close()
 
+  def keepalive(self):
+    async def send_ping():
+      if self._pong is False:
+        return await self.connection.close()
+      self._pong = False
+      await self.execute(Ping())
+      self._pong = True
+    asyncio.create_task(send_ping())
+    self._pingInterval = self.loop.call_later(self.PING_DELAY, self.keepalive)
+
   async def _clearExecuteQueue(self):
     while True:
       if self._executeQueue.empty():

signalwire/tests/blade/test_blade_messages.py

@@ -1,10 +1,11 @@
+from time import time
 from unittest import TestCase
-from unittest.mock import patch
 from signalwire import __version__
 from signalwire.blade.messages.message import Message
 from signalwire.blade.messages.connect import Connect
 from signalwire.blade.messages.execute import Execute
 from signalwire.blade.messages.subscription import Subscription
+from signalwire.blade.messages.ping import Ping
 
 class TestBladeMessages(TestCase):
   def test_from_json_with_result(self):
@@ -74,3 +75,18 @@ def test_subscription(self):
 
     self.assertEqual(msg.method, 'blade.subscription')
     self.assertEqual(msg.to_json(), '{"method":"blade.subscription","jsonrpc":"2.0","id":"mocked","params":{"protocol":"proto","command":"add","channels":["notif"]}}')
+
+  def test_ping_without_ts(self):
+    msg = Ping()
+    msg.id = 'mocked'
+
+    self.assertEqual(msg.method, 'blade.ping')
+    self.assertEqual(msg.to_json(), '{"method":"blade.ping","jsonrpc":"2.0","id":"mocked","params":{}}')
+
+  def test_ping_with_ts(self):
+    ts = time()
+    msg = Ping(ts)
+    msg.id = 'mocked'
+
+    self.assertEqual(msg.method, 'blade.ping')
+    self.assertEqual(msg.to_json(), f'{{"method":"blade.ping","jsonrpc":"2.0","id":"mocked","params":{{"timestamp":{ts}}}}}')