Quart: 如何使服务器推送事件（SSE）工作？

Question

我正在尝试在Quart中实现一个用于服务器推送事件（SSE）的接口，参考的是官方文档中的示例：http://pgjones.gitlab.io/quart/how_to_guides/server_sent_events.html

我复制粘贴了代码，并用一些假字符串作为data。结果导致我的服务器卡住了，因为它会不停地发送我的“数据”。所以这个示例并不能完全工作，你还得想办法正确地将事件添加到流中。

现在我使用asyncio.queues设置了一个合适的队列，当我添加东西时，send_events()函数现在可以响应了。太好了！

唯一的问题是，当我调用这个接口（用Postman）时，没有任何输出。它一直在等待响应。如果我在中途停止服务器，我会看到到那时为止生成的SSE输出。所以事件本身是被触发并且格式正确的，只是输出没有像应该那样被流式传输。

我发现了一个有相同问题的示例：https://github.com/encode/starlette/issues/20#issuecomment-586169195。不过，这个讨论的方向不同，最终为Starlette创建了一个实现。我尝试了这个包（我也在使用FastAPI），但是SSE接口的返回是EventSourceResponse，然后我遇到了错误：

TypeError: The response value type (EventSourceResponse) is not valid

对了...所以Quart不喜欢这个响应值。我看不出有什么办法能让它在Quart中工作，而且由于Quart文档中的示例不管用，看来唯一的选择就是放弃Quart了。

或者还有其他解决方案吗？

代码

我有一个数据类，里面只有event和data属性：

@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')

然后是接口 - 注意我加了一些打印语句，并且最开始在队列中添加了一些东西来测试（在代码的其他地方，事件会以同样的方式添加到队列中）：

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

异步编程 api接口流式传输 quart sse 事件流服务器推送数据队列

Quart: 如何使服务器推送事件（SSE）工作？

代码

1 个回答

撰写回答