fix incorrect schema

Author: BewareMyPower

pulsar/asyncio.py

@@ -85,14 +85,15 @@ def __init__(self, producer: _pulsar.Producer) -> None:
         """
         self._producer: _pulsar.Producer = producer
 
-    async def send(self, content: bytes) -> pulsar.MessageId:
+    async def send(self, content: Any) -> pulsar.MessageId:
         """
         Send a message asynchronously.
 
         parameters
         ----------
-        content: bytes
-            The message payload
+        content: Any
+            The message payload, whose type should respect the schema defined in
+            `Client.create_producer`.
 
         Returns
         -------
@@ -132,7 +133,7 @@ class Consumer:
     The Pulsar message consumer, used to subscribe to messages from a topic.
     """
 
-    def __init__(self, consumer: _pulsar.Consumer) -> None:
+    def __init__(self, consumer: _pulsar.Consumer, schema: pulsar.schema.Schema) -> None:
         """
         Create the consumer.
         Users should not call this constructor directly. Instead, create the
@@ -142,8 +143,11 @@ def __init__(self, consumer: _pulsar.Consumer) -> None:
         ----------
         consumer: _pulsar.Consumer
             The underlying Consumer object from the C extension.
+        schema: pulsar.schema.Schema
+            The schema of the data that will be received by this consumer.
         """
-        self._consumer: _pulsar.Consumer = consumer
+        self._consumer = consumer
+        self._schema = schema
 
     async def receive(self) -> pulsar.Message:
         """
@@ -163,7 +167,7 @@ async def receive(self) -> pulsar.Message:
         msg = await future
         m = pulsar.Message()
         m._message = msg
-        m._schema = pulsar.schema.BytesSchema()
+        m._schema = self._schema
         return m
 
     async def acknowledge(
@@ -633,7 +637,7 @@ async def subscribe(self, topic: Union[str, List[str]],
                 "(str, list)"
             )
 
-        return Consumer(await future)
+        return Consumer(await future, schema)
 
     async def close(self) -> None:
         """

tests/asyncio_test.py

@@ -252,6 +252,18 @@ async def test_seek_timestamp(self):
         msg = await consumer.receive()
         self.assertEqual(msg.data(), b'msg-3')
 
+    async def test_schema(self):
+        topic = f'asyncio-test-schema-{time.time()}'
+        producer = await self._client.create_producer(
+                topic, schema=pulsar.schema.StringSchema()
+        )
+        consumer = await self._client.subscribe(
+            topic, 'sub', schema=pulsar.schema.StringSchema()
+        )
+        await producer.send('test-message')
+        msg = await consumer.receive()
+        self.assertEqual(msg.value(), 'test-message')
+
 
 if __name__ == '__main__':
     main()