Change kick method to new interface

Author: chandr-andr

taskiq_aio_kafka/broker.py

@@ -183,12 +183,11 @@ async def kick(self, message: BrokerMessage) -> None:
         if not self._is_producer_started:
             raise ValueError("Please run startup before kicking.")
 
-        kafka_message: bytes = pickle.dumps(message)
         topic_name: str = self._kafka_topic.name
 
         await self._aiokafka_producer.send(
             topic=topic_name,
-            value=kafka_message,
+            value=message.message,
         )
 
     async def listen(

tests/test_broker.py

@@ -48,12 +48,7 @@ async def test_kick_success(broker: AioKafkaBroker) -> None:
         get_first_task(broker),
         timeout=1,
     )
-    assert pickle.dumps(message_to_send) == received_message_bytes
-
-    received_message: BrokerMessage = pickle.loads(
-        received_message_bytes,
-    )
-    assert message_to_send == received_message
+    assert received_message_bytes == message_to_send.message
 
 
 @pytest.mark.anyio
@@ -110,21 +105,12 @@ async def test_listen(
 
     await test_kafka_producer.send(
         topic=base_topic_name,
-        value=pickle.dumps(message_to_send),
+        value=message_to_send.message,
     )
 
     received_message_bytes: bytes = await asyncio.wait_for(
         get_first_task(broker),
         timeout=1,
     )
 
-    assert pickle.dumps(message_to_send) == received_message_bytes
-
-    received_message: BrokerMessage = pickle.loads(
-        received_message_bytes,
-    )
-
-    assert received_message.message == message
-    assert received_message.labels == labels
-    assert received_message.task_id == task_id
-    assert received_message.task_name == task_name
+    assert received_message_bytes == message_to_send.message