Hello,
This taskiq and faststream integration has been broken since taskiq version 0.11.1.
Run the following example using the command taskiq scheduler task_test:scheduler:
from faststream import FastStream
from faststream.confluent import KafkaBroker
broker = KafkaBroker("localhost:9092")
app = FastStream(broker)
@broker.subscriber("in-topic")
@broker.publisher("out-topic")
async def handle_msg(user: str, user_id: int) -> str:
msg = f"User: {user_id} - {user} registered"
print(msg)
return msg
from taskiq_faststream import BrokerWrapper
taskiq_broker = BrokerWrapper(broker)
from taskiq_faststream import StreamScheduler
from taskiq.schedule_sources import LabelScheduleSource
async def collect_information_to_send():
for i in range(10):
yield {"user": "John", "user_id": i}
taskiq_broker.task(
message=collect_information_to_send,
topic="in-topic",
schedule=[{
"cron": "* * * * *",
}],
)
scheduler = StreamScheduler(
broker=taskiq_broker,
sources=[LabelScheduleSource(taskiq_broker)],
)
This example works until taskiq version 0.11.0 but does not works starting from the taskiq version 0.11.1.
I believe taskiq's API changed and we need to update the integration.
The above example is taken from the faststream documentation page https://faststream.airt.ai/latest/scheduling/#generate-message-payload and the issue is initially raised at https://discord.com/channels/1085457301214855171/1085457302280228898/1275134622484795415
Hello,
This
taskiqandfaststreamintegration has been broken sincetaskiqversion0.11.1.Run the following example using the command
taskiq scheduler task_test:scheduler:This example works until taskiq version
0.11.0but does not works starting from the taskiq version0.11.1.I believe taskiq's API changed and we need to update the integration.
The above example is taken from the faststream documentation page https://faststream.airt.ai/latest/scheduling/#generate-message-payload and the issue is initially raised at https://discord.com/channels/1085457301214855171/1085457302280228898/1275134622484795415