如何同步处理响应代码时,如何异步处理龙卷风中的流数据？

Question

我的龙卷风API调用将调用另一个URL,然后将结果流回客户端.但是,如果内部URL返回错误代码,我想单独处理我自己的错误,并将错误内容流式传输到客户端.我现在拥有的是:

@web.asynchronous
@gen.coroutine
def get(self, job_id):
    url = ...
    client = httpclient.AsyncHTTPClient()

    # handle_chunk will forward received bytes to the client, allowing
    # other HTTP requests to be handled concurrently
    response = yield client.fetch(httpclient.HTTPRequest(
        url=url,
        streaming_callback=self._handle_chunk))
    self.finish()

def _handle_chunk(self, chunk):
    self.write(chunk)
    self.flush()

如果响应代码在200系列中,我需要将其修改为仅启动转发块,但client.fetch在整个请求完成之前不会产生响应.知道怎么做吗？

Answer 1

使用streaming_callback和header_callback.标头回调将在第一次调用之前运行streaming_callback,标头作为字符串列表.标题以状态行开头,您必须自己解析:

@gen.coroutine
def get(self, job_id):
    url = ...
    client = httpclient.AsyncHTTPClient()

    response = yield client.fetch(httpclient.HTTPRequest(
        url=url,
        header_callback=self._handle_headers,
        streaming_callback=self._handle_chunk))
    if response.code != 200:
        # handle error
    self.finish()

def _handle_headers(self, headers):
    self._start_line = tornado.httputil.parse_response_start_line(headers[0])

def _handle_chunk(self, chunk):
    if self._start_line.code == 200:
        self.write(chunk)
        self.flush()

目前无法彻底取消请求; 如果状态代码不是200,则回调将必须接受流式块并忽略它们(然后在其他地方返回错误).