Соединение закрыто потоком запросов и асинхронной обработкой

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