SocketDevice: notice dropped connections, and don't duplicate handlers on reopen

alarmdecoder/decoder.py

@@ -276,7 +276,7 @@ def open(self, baudrate=None, no_reader_thread=False):
             self._device.open(baudrate=baudrate,
                               no_reader_thread=no_reader_thread)
         except:
-            self._unwire_events
+            self._unwire_events()
             raise
 
         return self
@@ -393,6 +393,11 @@ def _wire_events(self):
         """
         Wires up the internal device events.
         """
+        # open() is called again to reconnect after the device closes, and the
+        # event handler lists don't de-duplicate; remove any earlier wiring so
+        # every message is still handled once.
+        self._unwire_events()
+
         self._device.on_open += self._on_open
         self._device.on_close += self._on_close
         self._device.on_read += self._on_read
@@ -402,14 +407,16 @@ def _wire_events(self):
 
     def _unwire_events(self):
         """
-        Wires up the internal device events.
+        Unwires the internal device events.  Safe to call when they aren't wired.
         """
-        self._device.on_open -= self._on_open
-        self._device.on_close -= self._on_close
-        self._device.on_read -= self._on_read
-        self._device.on_write -= self._on_write
-        self._zonetracker.on_fault -= self._on_zone_fault
-        self._zonetracker.on_restore -= self._on_zone_restore
+        for event, handler in ((self._device.on_open, self._on_open),
+                               (self._device.on_close, self._on_close),
+                               (self._device.on_read, self._on_read),
+                               (self._device.on_write, self._on_write),
+                               (self._zonetracker.on_fault, self._on_zone_fault),
+                               (self._zonetracker.on_restore, self._on_zone_restore)):
+            while handler in event:
+                event.remove(handler)
 
     def _handle_message(self, data):
         """

alarmdecoder/devices/socket_device.py

@@ -188,6 +188,7 @@ def open(self, baudrate=None, no_reader_thread=False):
             self._read_thread = Device.ReadThread(self)
 
             self._device = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self._enable_keepalive()
 
             if self._use_ssl:
                 self._init_ssl()
@@ -327,7 +328,13 @@ def timeout_event():
 
                 buf = self._device.recv(1)
 
-                if buf != b'' and buf != b"\xff":
+                # An empty read on a readable socket means the remote end
+                # closed the connection.  Ignoring it would leave the
+                # reader spinning on a dead socket, and on_close would never fire.
+                if buf == b'':
+                    raise CommError('Connection closed by {0}:{1}'.format(self._host, self._port))
+
+                if buf != b"\xff":
                     ub = bytes_hack(buf)
 
                     self._buffer += ub
@@ -363,6 +370,24 @@ def timeout_event():
 
         return ret.decode('utf-8')
 
+    def _enable_keepalive(self):
+        """
+        Turns on TCP keepalive so that a peer that vanished without closing the
+        connection is eventually noticed.  Best effort: options the platform
+        doesn't have are skipped.
+        """
+        try:
+            self._device.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+
+            # macOS has no TCP_KEEPIDLE; its TCP_KEEPALIVE is the same setting.
+            idle = 'TCP_KEEPIDLE' if hasattr(socket, 'TCP_KEEPIDLE') else 'TCP_KEEPALIVE'
+            for option, value in ((idle, 30), ('TCP_KEEPINTVL', 10), ('TCP_KEEPCNT', 3)):
+                if hasattr(socket, option):
+                    self._device.setsockopt(socket.IPPROTO_TCP, getattr(socket, option), value)
+
+        except (socket.error, AttributeError):
+            pass
+
     def purge(self):
         """
         Purges read/write buffers.

test/test_ad2.py

@@ -148,6 +148,23 @@ def test_close(self):
         self._decoder.close()
         self._device.close.assert_called()
 
+    def test_reopen_does_not_duplicate_handlers(self):
+        self._decoder.open()
+        self._decoder.open()  # e.g. reconnecting after the device closed
+
+        self.assertEqual(list(self._device.on_read).count(self._decoder._on_read), 1)
+        self.assertEqual(list(self._device.on_close).count(self._decoder._on_close), 1)
+
+    def test_failed_open_unwires_events(self):
+        self._device.open.side_effect = Exception('connection refused')
+
+        for _ in range(3):
+            with self.assertRaises(Exception):
+                self._decoder.open()
+
+        self.assertNotIn(self._decoder._on_read, list(self._device.on_read))
+        self.assertNotIn(self._decoder._on_close, list(self._device.on_close))
+
     def test_send(self):
         self._decoder.send('test')
         self._device.write.assert_called_with(b'test')

test/test_devices.py

@@ -228,6 +228,31 @@ def test_read_line_exception(self):
                     with self.assertRaises(CommError):
                         self._device.read_line()
 
+    def test_read_line_remote_closed(self):
+        with patch.object(select, 'select', return_value=[[1], [], []]):
+            with patch.object(self._device._device, 'recv', return_value=b''):
+                with self.assertRaises(CommError):
+                    self._device.read_line(timeout=1)
+
+    def test_open_enables_keepalive(self):
+        with patch.object(socket.socket, '__init__', return_value=None):
+            with patch.object(socket.socket, 'connect', return_value=None):
+                with patch.object(socket.socket, 'setsockopt') as setsockopt:
+                    self._device.open(no_reader_thread=True)
+
+        setsockopt.assert_any_call(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+
+    def test_keepalive_idle_without_tcp_keepidle(self):
+        # macOS has TCP_KEEPALIVE (the idle time) instead of TCP_KEEPIDLE.
+        sock = Mock()
+        self._device._device = sock
+
+        with patch.dict(socket.__dict__, {'TCP_KEEPALIVE': 0x10}):
+            socket.__dict__.pop('TCP_KEEPIDLE', None)
+            self._device._enable_keepalive()
+
+        sock.setsockopt.assert_any_call(socket.IPPROTO_TCP, 0x10, 30)
+
     def test_ssl(self):
         if not have_openssl:
             return