From 11fd3659d1960a5e4b7a35442231b7b4fcccce48 Mon Sep 17 00:00:00 2001 From: Paul Traina Date: Wed, 21 Oct 2026 10:06:24 -0700 Subject: [PATCH] SocketDevice: notice dropped connections; don't duplicate handlers on reopen read_line() ignored an empty recv(), so when ser2sock or the network closed the connection the reader spun on a dead socket and on_close never fired. Anything reconnecting on on_close (Home Assistant does) never got the chance. An empty read now raises CommError, which the reader thread already turns into close() and on_close. A link that dies without a close (Wi-Fi drop, a firewall flushing its state table) can't be seen from a read-only socket, so TCP keepalive is now on: 30 s idle, then 3 probes 10 s apart (best effort; options the platform doesn't have are skipped, and on macOS the idle time is TCP_KEEPALIVE). A dead link is noticed in about a minute. AlarmDecoder.open() wired the device events again on every call, and the handler lists don't de-duplicate, so every reconnect doubled message handling. The failed-open path meant to undo it but was missing the (). Wiring is now idempotent and a failed open unwires. --- alarmdecoder/decoder.py | 23 +++++++++++++++-------- alarmdecoder/devices/socket_device.py | 27 ++++++++++++++++++++++++++- test/test_ad2.py | 17 +++++++++++++++++ test/test_devices.py | 25 +++++++++++++++++++++++++ 4 files changed, 83 insertions(+), 9 deletions(-) diff --git a/alarmdecoder/decoder.py b/alarmdecoder/decoder.py index 00e70fd..8195f25 100644 --- a/alarmdecoder/decoder.py +++ b/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): """ diff --git a/alarmdecoder/devices/socket_device.py b/alarmdecoder/devices/socket_device.py index 9feced4..7949ec0 100644 --- a/alarmdecoder/devices/socket_device.py +++ b/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. diff --git a/test/test_ad2.py b/test/test_ad2.py index 34feb84..dfb5126 100644 --- a/test/test_ad2.py +++ b/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') diff --git a/test/test_devices.py b/test/test_devices.py index d418780..5f2db28 100644 --- a/test/test_devices.py +++ b/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