Quart中Server Sent Events(SSE)无法正常流式输出的问题求助
Quart中实现Server Sent Events(SSE)端点的问题
尝试按照Quart官方文档实现SSE端点,复制示例代码并加入测试数据后,服务器因无限流式输出该测试数据直接锁死,说明官方示例无法直接运行,必须自行实现事件添加逻辑。
改用asyncio.queues搭建队列后,send_events()函数已能正常响应事件添加,但调用该端点时(通过Postman)始终没有输出,一直处于等待响应状态;只有中途停止服务器,才能获取到截至当时生成的SSE内容——这说明事件已触发且格式正确,但输出没有按预期进行流式传输。
曾找到过类似问题的讨论,但最终方案是针对Starlette框架实现的。尝试使用相关包(同时我也在使用FastAPI),但在SSE端点返回EventSourceResponse时出现错误:
TypeError: The response value type (EventSourceResponse) is not valid
显然Quart不支持该响应类型,而官方示例又无法正常运行,难道只能放弃Quart?有没有其他可行的解决方案?
代码
数据类定义
@dataclass class ServerSentEvent: data: str event: str def encode(self) -> bytes: message = f'data: {self.data}' message = f'{message}\nevent: {self.event}' message = f'{message}\r\n\r\n' return message.encode('utf-8')
SSE端点实现
queues = [] @bp.get('/sse') async def sse(): """Server Sent Events""" if 'text/event-stream' not in request.accept_mimetypes: abort(400) async def send_events(): while True: print('waiting for event') event = await queue.get() print('got event') print(event.encode()) yield event.encode() queue = Queue() queues.append(queue) await queue.put(ServerSentEvent(**{'event': 'subscribed', 'data': 'will send new events'})) print('-- SSE connected') response = await make_response( send_events(), { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Transfer-Encoding': 'chunked', 'Connection': 'keep-alive' } ) response.timeout = None return response
内容的提问来源于stack exchange,提问作者kasimir
相关产品推荐
相关产品推荐

