使用请求流和异步处理关闭连接

Question

我在处理来自 HTTP 请求的流时遇到问题。处理程序从流中收集数据并在发送响应之前执行一些异步操作。

一旦收集到流数据，HTTP 连接就会在someAsyncStuff 被调用之前关闭。

这是重现问题的示例。我做错了什么？

import * as Busboy from 'busboy'
import * as express from 'express'
import { pipeline } from 'stream'
import concat = require('concat-stream')

const app = express()

app.post('/', (req, res) => {
  const busboy = new Busboy({ headers: req.headers })

  busboy.on('file', (_, file) => {
    pipeline(
      file,
      concat(buffer =>
        someAsyncStuff(buffer.toString())
          .then(length => res.send({ length }))
          .catch(err => res.status(500).send(err.message))
      ),
      err => {
        if (err) res.status(500).send(err.message)
      }
    )
  })

  pipeline(req, busboy, err => {
    if (err) res.status(500).send(err.message)
  })
})

function someAsyncStuff(s: string): Promise<number> {
  return new Promise(resolve => setTimeout(() => resolve(s.length), 1))
}

app.listen('3000')

Answer 1

使用 req.pipe 似乎有效。

import Busboy from 'busboy'
import express from 'express'
import { pipeline } from 'stream'
import concat from 'concat-stream'

const app = express()

app.post('/', (req, res) => {
  const busboy = new Busboy({ headers: req.headers })

  busboy.on('file', (_, file) => {
    pipeline(
      file,
      concat(buffer =>
        someAsyncStuff(buffer.toString())
          .then(length => res.json({ length }))
          .catch(err => res.status(500).send(err.message))
      ),
      err => {
        if (err) res.status(500).send(err.message)
      }
    )
  })
  req.pipe(busboy);
})

function someAsyncStuff(s) {
  return new Promise(resolve => setTimeout(() => resolve(s.length), 1000))
}

app.listen('3000')