File upload with Node streams

Question

I'm implementing a file upload using NodeJS. I've got fully functional code, however, it feels overly complicated and I'm hoping there would be an easier way to implement the same thing?

In particular what I want to achieve is:

• Write the request to a file
• Send upload progress information
• Send final MD5 hash
• Send final size (position)
• Properly propagate errors
• Have no resource leaks

---

app.post('/file', (req, res, next) => {

  // ...

  const progress$ = Observable
    .create(o => {
      const fileStream = fs.createWriteStream(filePath, { flags: 'wx' })
      const hasher = crypto.createHash('md5')

      req.on('data', buf => {
        hasher.update(buf)
        o.next({
          position: fileStream.bytesWritten
        })
        if (!fileStream.write(buf)) {
          req.pause()
        }
      })
      req.on('error', err => {
        fileStream.destroy()
        o.error(err)
      })
      req.on('end', ::fileStream.end)
      req.on('close', () => {
        fileStream.destroy()
        o.error(new Error('upload cancelled'))
      })
      fileStream.on('drain', ::req.resume)
      fileStream.on('error', ::o.error)
      fileStream.on('close', () => {
        o.next({
          md5: hasher.digest('hex'),
          position: fileStream.bytesWritten
        })
        o.complete()
      })
    })

 // ...

})