Skip to content

Commit 1dba609

Browse files
emaneman
authored
Merge pull request #88 from eman/fix/mqtt-resubscribe-on-session-lost
fix: resubscribe all topics when session_present=False on connection …
2 parents 231624f + 8e91ebb commit 1dba609

3 files changed

Lines changed: 227 additions & 8 deletions

File tree

‎src/nwp500/mqtt/client.py‎

Lines changed: 37 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -364,8 +364,20 @@ def _on_connection_resumed_internal(
364364
)
365365
)
366366

367-
# Send any queued commands
368-
if self.config.enable_command_queue and self._command_queue:
367+
# When the broker starts a clean session (session_present=False), all
368+
# previous subscriptions have been dropped server-side. We must
369+
# re-establish them before any device data can flow. This covers the
370+
# common case where the AWS IoT SDK auto-reconnects internally before
371+
# the MqttReconnectionHandler fires its own reconnect path — in that
372+
# scenario the reconnect handler sees _connected==True and exits early,
373+
# so resubscribe_all() would never be called without this block.
374+
#
375+
# When session_present=False, we must resubscribe before sending queued
376+
# commands to ensure subscriptions are restored before device responses
377+
# are processed. Use a composite coroutine to enforce ordering.
378+
if not session_present and self._subscription_manager:
379+
self._schedule_coroutine(self._handle_clean_session_resume())
380+
elif self.config.enable_command_queue and self._command_queue:
369381
self._schedule_coroutine(self._send_queued_commands_internal())
370382

371383
async def _send_queued_commands_internal(self) -> None:
@@ -377,6 +389,29 @@ async def _send_queued_commands_internal(self) -> None:
377389
self._connection_manager.publish, lambda: self._connected
378390
)
379391

392+
async def _handle_clean_session_resume(self) -> None:
393+
"""
394+
Handle clean session reconnection with ordered resubscription.
395+
396+
When session_present=False (clean session), the broker has dropped all
397+
subscriptions. This method ensures subscriptions are restored BEFORE
398+
sending any queued commands, preventing commands from being processed
399+
before their subscriptions are re-established.
400+
"""
401+
if not self._subscription_manager or not self._connection_manager:
402+
return
403+
404+
if not self._connection_manager.connection:
405+
return
406+
407+
self._subscription_manager.update_connection(
408+
self._connection_manager.connection
409+
)
410+
await self._subscription_manager.resubscribe_all()
411+
412+
if self.config.enable_command_queue and self._command_queue:
413+
await self._send_queued_commands_internal()
414+
380415
async def _active_reconnect(self) -> None:
381416
"""
382417
Actively trigger a reconnection attempt.

‎src/nwp500/mqtt/periodic.py‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -173,12 +173,6 @@ async def periodic_request() -> None:
173173
await self._request_device_info(device)
174174
elif request_type == PeriodicRequestType.DEVICE_STATUS:
175175
await self._request_device_status(device)
176-
else:
177-
_logger.error(
178-
"Unknown periodic request type: %s",
179-
request_type,
180-
)
181-
break
182176

183177
_logger.debug(
184178
"Sent periodic %s request for %s",

‎tests/test_mqtt_clean_session_resume.py‎

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
1+
"""Tests for MQTT client clean session reconnection handling."""
2+
3+
from __future__ import annotations
4+
5+
from unittest.mock import AsyncMock, MagicMock, patch
6+
7+
import pytest
8+
9+
from nwp500.auth import AuthenticationResponse, AuthTokens, UserInfo
10+
from nwp500.mqtt import NavienMqttClient
11+
12+
13+
@pytest.fixture
14+
def auth_client_with_valid_tokens():
15+
"""Create an auth client with valid tokens."""
16+
from nwp500.auth import NavienAuthClient
17+
18+
auth_client = NavienAuthClient("test@example.com", "password")
19+
valid_tokens = AuthTokens(
20+
id_token="test_id",
21+
access_token="test_access",
22+
refresh_token="test_refresh",
23+
authentication_expires_in=3600,
24+
access_key_id="test_key_id",
25+
secret_key="test_secret_key",
26+
session_token="test_session",
27+
authorization_expires_in=3600,
28+
)
29+
auth_client._auth_response = AuthenticationResponse(
30+
user_info=UserInfo(user_first_name="Test", user_last_name="User"),
31+
tokens=valid_tokens,
32+
)
33+
return auth_client
34+
35+
36+
class TestMqttCleanSessionResume:
37+
"""Tests for clean session (session_present=False) reconnection handling."""
38+
39+
@pytest.mark.asyncio(loop_scope="function")
40+
async def test_on_connection_resumed_with_clean_session_resubscribes(
41+
self, auth_client_with_valid_tokens
42+
):
43+
"""Resubscribe when session_present=False on connection resume."""
44+
client = NavienMqttClient(auth_client_with_valid_tokens)
45+
46+
# Mock the components
47+
mock_subscription_manager = AsyncMock()
48+
mock_subscription_manager.resubscribe_all = AsyncMock()
49+
client._subscription_manager = mock_subscription_manager
50+
51+
mock_connection_manager = MagicMock()
52+
mock_connection = MagicMock()
53+
mock_connection_manager.connection = mock_connection
54+
client._connection_manager = mock_connection_manager
55+
56+
# Mock the event emitter and diagnostics
57+
client.emit = AsyncMock()
58+
client._diagnostics = MagicMock()
59+
client._diagnostics.record_connection_success = AsyncMock()
60+
61+
# Call with session_present=False (clean session)
62+
client._on_connection_resumed_internal(
63+
connection=mock_connection, return_code=0, session_present=False
64+
)
65+
66+
# Give the scheduled coroutine time to run
67+
import asyncio
68+
69+
await asyncio.sleep(0.1)
70+
71+
# Verify resubscribe_all was called
72+
mock_subscription_manager.update_connection.assert_called_once_with(
73+
mock_connection
74+
)
75+
# The resubscribe should be scheduled via _schedule_coroutine
76+
# We need to wait for it or check the internal state
77+
78+
@pytest.mark.asyncio(loop_scope="function")
79+
async def test_resubscribe_before_queued_commands(
80+
self, auth_client_with_valid_tokens
81+
):
82+
"""Resubscribe completes before queued commands are sent."""
83+
client = NavienMqttClient(auth_client_with_valid_tokens)
84+
85+
# Track call order
86+
call_order = []
87+
88+
# Mock the components
89+
mock_subscription_manager = MagicMock()
90+
mock_subscription_manager.resubscribe_all = AsyncMock(
91+
side_effect=lambda: call_order.append("resubscribe")
92+
)
93+
client._subscription_manager = mock_subscription_manager
94+
95+
mock_connection_manager = MagicMock()
96+
mock_connection = MagicMock()
97+
mock_connection_manager.connection = mock_connection
98+
client._connection_manager = mock_connection_manager
99+
100+
# Mock command queue
101+
client._command_queue = AsyncMock()
102+
client.config.enable_command_queue = True
103+
104+
# Mock send_queued_commands to track it's called after resubscribe
105+
original_send = client._send_queued_commands_internal
106+
107+
async def mock_send():
108+
call_order.append("send_queued")
109+
await original_send()
110+
111+
client._send_queued_commands_internal = mock_send
112+
113+
# Call the method
114+
await client._handle_clean_session_resume()
115+
116+
# Verify subscription manager was updated with connection
117+
mock_subscription_manager.update_connection.assert_called_once_with(
118+
mock_connection
119+
)
120+
121+
# Verify resubscribe was called before queued commands
122+
assert call_order == ["resubscribe", "send_queued"]
123+
124+
@pytest.mark.asyncio(loop_scope="function")
125+
async def test_skip_when_no_subscription_manager(
126+
self, auth_client_with_valid_tokens
127+
):
128+
"""Return early if subscription_manager is None."""
129+
client = NavienMqttClient(auth_client_with_valid_tokens)
130+
client._subscription_manager = None
131+
132+
# Should not raise
133+
await client._handle_clean_session_resume()
134+
135+
@pytest.mark.asyncio(loop_scope="function")
136+
async def test_handle_clean_session_resume_skips_when_no_connection(
137+
self, auth_client_with_valid_tokens
138+
):
139+
"""Return early if connection is None."""
140+
client = NavienMqttClient(auth_client_with_valid_tokens)
141+
142+
mock_subscription_manager = MagicMock()
143+
client._subscription_manager = mock_subscription_manager
144+
145+
mock_connection_manager = MagicMock()
146+
mock_connection_manager.connection = None
147+
client._connection_manager = mock_connection_manager
148+
149+
# Should not raise
150+
await client._handle_clean_session_resume()
151+
152+
# Should not try to update connection
153+
mock_subscription_manager.update_connection.assert_not_called()
154+
155+
@pytest.mark.asyncio(loop_scope="function")
156+
async def test_on_connection_resumed_with_session_sends_queued_commands(
157+
self, auth_client_with_valid_tokens
158+
):
159+
"""Send queued commands normally when session_present=True."""
160+
client = NavienMqttClient(auth_client_with_valid_tokens)
161+
162+
# Mock the components
163+
mock_command_queue = AsyncMock()
164+
client._command_queue = mock_command_queue
165+
client.config.enable_command_queue = True
166+
167+
# Mock the event emitter and diagnostics
168+
client.emit = AsyncMock()
169+
client._diagnostics = MagicMock()
170+
client._diagnostics.record_connection_success = AsyncMock()
171+
172+
# Mock connection
173+
mock_connection = MagicMock()
174+
175+
# Patch _send_queued_commands_internal to track if called
176+
with patch.object(
177+
client, "_send_queued_commands_internal", new_callable=AsyncMock
178+
):
179+
# Call with session_present=True (session resumed)
180+
client._on_connection_resumed_internal(
181+
connection=mock_connection, return_code=0, session_present=True
182+
)
183+
184+
# Give the scheduled coroutine time to run
185+
import asyncio
186+
187+
await asyncio.sleep(0.1)
188+
189+
# Verify send_queued_commands_internal was scheduled
190+
# (it will be called through _schedule_coroutine)

0 commit comments

Comments
 (0)