feat: allow seeding the monitor resume position

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)

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()))

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:<epochMillis> <regex>`: 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()