diff --git a/at_client/atclient.py b/at_client/atclient.py index 32a6227..7f0feac 100644 --- a/at_client/atclient.py +++ b/at_client/atclient.py @@ -373,14 +373,16 @@ def __del__(self): if self.secondary_connection: self.secondary_connection.disconnect() - def start_monitor(self, regex=".*"): + def start_monitor(self, regex=".*", last_received_time=0): if self.queue != None: global should_be_running_lock what = "" try: if self.monitor_connection == None: what = "construct an AtMonitorConnection" - self.monitor_connection = AtMonitorConnection(queue=self.queue, atsign=self.atsign, address=self.secondary_address, verbose=self.verbose, regex=regex) + self.monitor_connection = AtMonitorConnection( + queue=self.queue, atsign=self.atsign, address=self.secondary_address, + verbose=self.verbose, regex=regex, last_received_time=last_received_time) self.monitor_connection.connect() AuthUtil.authenticate_with_pkam(self.monitor_connection, self.atsign, self.keys) should_be_running_lock.acquire(blocking=1) diff --git a/at_client/connections/atmonitorconnection.py b/at_client/connections/atmonitorconnection.py index a0d61ca..737ea95 100644 --- a/at_client/connections/atmonitorconnection.py +++ b/at_client/connections/atmonitorconnection.py @@ -15,14 +15,16 @@ import queue class AtMonitorConnection(AtSecondaryConnection): - last_received_time: int = 0 running: bool = False should_be_running: bool = False - - def __init__(self, queue:queue.Queue, atsign:AtSign, address: Address, context:ssl.SSLContext=ssl.create_default_context(), verbose:bool=True, regex=".*"): + + def __init__(self, queue: queue.Queue, atsign: AtSign, address: Address, + context: ssl.SSLContext = ssl.create_default_context(), + verbose: bool = True, regex=".*", last_received_time: int = 0): self.atsign = atsign self.queue = queue self.regex = regex + self.last_received_time = last_received_time self._verbose = verbose super().__init__(address, context, verbose) self._last_heartbeat_sent_time = TimeUtil.current_time_millis() @@ -123,11 +125,15 @@ def _is_shared_key_notification(atsign, key): """ return key.startswith(atsign.to_string() + ":shared_key@") + def _build_monitor_command(self): + """Monitor verb requesting notifications received after last_received_time.""" + return "monitor:" + str(self.last_received_time) + " " + self.regex + def _run(self): what = "" first = True try: - monitor_cmd = "monitor:" + str(self.last_received_time) + " " + self.regex + monitor_cmd = self._build_monitor_command() what = "send monitor command " + monitor_cmd self.execute_command(command=monitor_cmd, retry_on_exception=True, read_the_response=False) print("Monitor started on " + str(self.atsign.to_string())) diff --git a/test/monitor_resume_test.py b/test/monitor_resume_test.py new file mode 100644 index 0000000..ab05e03 --- /dev/null +++ b/test/monitor_resume_test.py @@ -0,0 +1,51 @@ +import unittest +from queue import Queue +from unittest.mock import patch + +from at_client import AtClient +from at_client.common import AtSign +from at_client.connections.address import Address +from at_client.connections.atmonitorconnection import AtMonitorConnection + + +class MonitorResumeTest(unittest.TestCase): + """Network-free tests for seeding the monitor resume position. + + The monitor verb is `monitor: `: the server replays + notifications received after epochMillis. Before this change + last_received_time could not be seeded, so a caller that rebuilt its client + after a dropped monitor always sent `monitor:0` instead of resuming. + """ + + def _connection(self, **kwargs): + with patch.object(AtMonitorConnection, "start_heart_beat"): + return AtMonitorConnection(queue=Queue(), atsign=AtSign("@alice"), + address=Address("localhost", 64), + verbose=False, regex="test", **kwargs) + + def test_default_starts_from_zero(self): + conn = self._connection() + self.assertEqual(conn.last_received_time, 0) + self.assertEqual(conn._build_monitor_command(), "monitor:0 test") + + def test_seeded_position_is_used_in_monitor_command(self): + conn = self._connection(last_received_time=1720000000000) + self.assertEqual(conn._build_monitor_command(), "monitor:1720000000000 test") + + def test_atclient_start_monitor_passes_seed_through(self): + client = AtClient.__new__(AtClient) # bypass the network-connecting __init__ + client.queue = Queue() + client.monitor_connection = None + client.atsign = AtSign("@alice") + client.secondary_address = Address("localhost", 64) + client.verbose = False + client.keys = {} + with patch("at_client.atclient.AtMonitorConnection") as connection_cls, \ + patch("at_client.atclient.AuthUtil.authenticate_with_pkam"): + connection_cls.return_value.running = True # skip start_monitor() on the mock + client.start_monitor("test", last_received_time=42) + self.assertEqual(connection_cls.call_args.kwargs["last_received_time"], 42) + + +if __name__ == "__main__": + unittest.main()