Add for_each_and_listen

Author: BewareMyPower

pulsar/tableview.py

@@ -66,6 +66,19 @@ def for_each(self, callback: Callable[[str, Any], None]) -> None:
         """
         self._table_view.for_each(lambda k, v: callback(k, self._schema.decode(v)))
 
+    def for_each_and_listen(self, callback: Callable[[str, Any], None]) -> None:
+        """
+        Iterate over all entries in the table view and call the callback function
+        with the key and value for each entry, then listen for changes. The callback
+        will be called when a new entry is added or an existing entry is updated.
+
+        Parameters
+        ----------
+        callback: Callable[[str, Any], None]
+            The callback function to call for each entry.
+        """
+        self._table_view.for_each_and_listen(lambda k, v: callback(k, self._schema.decode(v)))
+
     def close(self) -> None:
         """
         Close the table view.

src/table_view.cc

@@ -52,6 +52,7 @@ void export_table_view(py::module_& m) {
              })
         .def("size", &TableView::size, py::call_guard<py::gil_scoped_release>())
         .def("for_each", &TableView::forEach, py::call_guard<py::gil_scoped_release>())
+        .def("for_each_and_listen", &TableView::forEachAndListen, py::call_guard<py::gil_scoped_release>())
         .def("close", [](TableView& view) {
             waitForAsyncResult([&view](ResultCallback callback) { view.closeAsync(callback); });
         });

tests/table_view_test.py

@@ -66,6 +66,29 @@ def test_for_each(self):
             'key-1': 'value-1'
         })
 
+        def listener(key: str, value: str):
+            if len(value) == 0:
+                d.pop(key)
+            else:
+                d[key] = value
+
+        d.clear()
+        table_view.for_each_and_listen(listener)
+        self.assertEqual(d, {
+            'key-0': 'value-0',
+            'key-1': 'value-1'
+        })
+
+        producer.send('value-0-new'.encode(), partition_key='key-0')
+        producer.send(''.encode(), partition_key='key-1')
+        producer.send('value-2'.encode(), partition_key='key-2')
+        def assert_latest_values():
+            self.assertEqual(d, {
+                'key-0': 'value-0-new',
+                'key-2': 'value-2'
+            })
+        self._wait_for_assertion(assert_latest_values)
+
     def _wait_for_assertion(self, assertion: Callable, timeout=5) -> None:
         start_time = time.time()
         while time.time() - start_time < timeout: