Support for_each

Author: BewareMyPower

pulsar/tableview.py

@@ -21,7 +21,7 @@
 The TableView implementation.
 """
 
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 from pulsar.schema.schema import Schema
 import _pulsar
 
@@ -53,10 +53,18 @@ def get(self, key: str) -> Optional[Any]:
             return self._schema.decode(pair[1])
         else:
             return None
-        #value = self._table_view.get(key)
-        #if value is None:
-        #    return None
-        #return self._schema.decode(value)
+
+    def for_each(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.
+
+        Parameters
+        ----------
+        callback: Callable[[str, Any], None]
+            The callback function to call for each entry.
+        """
+        self._table_view.for_each(lambda k, v: callback(k, self._schema.decode(v)))
 
     def close(self) -> None:
         """

src/table_view.cc

@@ -21,6 +21,7 @@
 #include <pulsar/Schema.h>
 #include <pulsar/TableViewConfiguration.h>
 #include <pybind11/stl.h>
+#include <pybind11/functional.h>
 #include <utility>
 #include "utils.h"
 
@@ -50,6 +51,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("close", [](TableView& view) {
             waitForAsyncResult([&view](ResultCallback callback) { view.closeAsync(callback); });
         });

tests/table_view_test.py

@@ -29,7 +29,7 @@ class TableViewTest(TestCase):
     def setUp(self):
         self._client: Client = Client('pulsar://localhost:6650')
 
-    def tearDown(self) -> None:
+    def tearDown(self):
         self._client.close()
 
     def test_get(self):
@@ -51,6 +51,20 @@ def test_get(self):
         producer.close()
         table_view.close()
 
+    def test_for_each(self):
+        topic = f'table_view_test_for_each-{time.time()}'
+        table_view = self._client.create_table_view(topic)
+        producer = self._client.create_producer(topic)
+        producer.send('value-0'.encode(), partition_key='key-0')
+        producer.send('value-1'.encode(), partition_key='key-1')
+        self._wait_for_assertion(lambda: self.assertEqual(len(table_view), 2))
+
+        d = dict()
+        table_view.for_each(lambda key, value: d.__setitem__(key, value))
+        self.assertEqual(d, {
+            'key-0': 'value-0',
+            'key-1': 'value-1'
+        })
 
     def _wait_for_assertion(self, assertion: Callable, timeout=5) -> None:
         start_time = time.time()