Fix schema not respected

Author: BewareMyPower

src/table_view.cc

@@ -22,6 +22,7 @@
 #include <pulsar/TableViewConfiguration.h>
 #include <pybind11/stl.h>
 #include <pybind11/functional.h>
+#include <functional>
 #include <utility>
 #include "utils.h"
 
@@ -51,8 +52,22 @@ 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("for_each",
+             [](TableView& view, std::function<void(std::string, py::bytes)> callback) {
+                 py::gil_scoped_release release;
+                 view.forEach([callback](const std::string& key, const std::string& value) {
+                     py::gil_scoped_acquire acquire;
+                     callback(key, py::bytes(value));
+                 });
+             })
+        .def("for_each_and_listen",
+             [](TableView& view, std::function<void(std::string, py::bytes)> callback) {
+                 py::gil_scoped_release release;
+                 view.forEachAndListen([callback](const std::string& key, const std::string& value) {
+                     py::gil_scoped_acquire acquire;
+                     callback(key, py::bytes(value));
+                 });
+             })
         .def("close", [](TableView& view) {
             waitForAsyncResult([&view](ResultCallback callback) { view.closeAsync(callback); });
         });

tests/table_view_test.py

@@ -23,6 +23,7 @@
 import time
 
 from pulsar import Client
+from pulsar.schema.schema import StringSchema
 
 class TableViewTest(TestCase):
 
@@ -38,14 +39,14 @@ def test_get(self):
         self.assertEqual(len(table_view), 0)
 
         producer = self._client.create_producer(topic)
-        producer.send('value-0'.encode(), partition_key='key-0')
+        producer.send(b'value-0', partition_key='key-0')
         producer.send(b'\xba\xd0\xba\xd0', partition_key='key-1') # an invalid UTF-8 bytes
 
         self._wait_for_assertion(lambda: self.assertEqual(len(table_view), 2))
         self.assertEqual(table_view.get('key-0'), b'value-0')
         self.assertEqual(table_view.get('key-1'), b'\xba\xd0\xba\xd0')
 
-        producer.send('value-1'.encode(), partition_key='key-0')
+        producer.send(b'value-1', partition_key='key-0')
         self._wait_for_assertion(lambda: self.assertEqual(table_view.get('key-0'), b'value-1'))
 
         producer.close()
@@ -55,15 +56,15 @@ 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')
+        producer.send(b'value-0', partition_key='key-0')
+        producer.send(b'value-1', 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'
+            'key-0': b'value-0',
+            'key-1': b'value-1'
         })
 
         def listener(key: str, value: str):
@@ -75,20 +76,40 @@ def listener(key: str, value: str):
         d.clear()
         table_view.for_each_and_listen(listener)
         self.assertEqual(d, {
-            'key-0': 'value-0',
-            'key-1': 'value-1'
+            'key-0': b'value-0',
+            'key-1': b'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')
+        producer.send(b'value-0-new', partition_key='key-0')
+        producer.send(b'', partition_key='key-1')
+        producer.send(b'value-2', partition_key='key-2')
         def assert_latest_values():
             self.assertEqual(d, {
-                'key-0': 'value-0-new',
-                'key-2': 'value-2'
+                'key-0': b'value-0-new',
+                'key-2': b'value-2'
             })
         self._wait_for_assertion(assert_latest_values)
 
+    def test_schema(self):
+        topic = f'table_view_test_schema-{time.time()}'
+        table_view = self._client.create_table_view(topic, schema=StringSchema())
+        producer = self._client.create_producer(topic, schema=StringSchema())
+        producer.send('value', partition_key='key')
+
+        self._wait_for_assertion(lambda: self.assertEqual(table_view.get('key'), 'value'))
+        self.assertEqual(table_view.get('missed-key'), None)
+
+        entries = dict()
+        table_view.for_each(lambda key, value: entries.__setitem__(key, value))
+        self.assertEqual(entries, {'key': 'value'})
+
+        entries.clear()
+        table_view.for_each_and_listen(lambda key, value: entries.__setitem__(key, value))
+        self.assertEqual(entries, {'key': 'value'})
+
+        producer.send('new-value', partition_key='key')
+        self._wait_for_assertion(lambda: self.assertEqual(table_view.get('key'), 'new-value'))
+
     def _wait_for_assertion(self, assertion: Callable, timeout=5) -> None:
         start_time = time.time()
         while time.time() - start_time < timeout: