CODEWITHSUNDEEP
javascriptnode.jstypescripteventsnodejs-stream
Federico Morrone
Federico Morrone

Reputation: 83

How to make a function with events and pipe in nodejs

I'm writing a nodejs library in typescript, the main scope of this library is gonna be downloading stuff from a given url, what I'd like it to do is to be used like this

import library from 'library'

library('https://www.example.com')
    .on('progress', (progress: progress) => {
        //do something with the progress
    })
    .on('end', () => {
        //do something when done
    })
    .pipe(fs.createWriteStream('./test/file.mp4'))

I've never worked with node events and stream in such a way so I have no idea how to even go about this I'm using typescript and webpack also please forgive bad English

Upvotes: 2

Views: 1980

Answers (1)

shkaper
shkaper

Reputation: 4998

You're going to need to implement a Readable stream. Node streams are instances of EventEmitter, so you automatically have access to the event API.

At the very minimum, you need to implement a _read() method which is called internally whenever the consumer is ready to receive more data from the queue. Since you want the library to report the progress, you also need to keep track of how much data has been processed and emit events accordingly.

The code below ignores several important things like backpressuring, but it's a start. I'm using node-fetch as a request library as it exposes an underlying response stream and is fairly easy to use.

// fileLoader.js
const {Readable} = require('stream')
const fetch = require('node-fetch')

class FileLoader extends Readable {
  constructor(url) {
    super()
    this._url = url
    this._fetchStarted = false
    this._totalLength = 0
    this._currentLength = 0
  }

  _processData(stream) {
    stream
      .on('end', () => {
        this.push(null)
      })
      .on('error', (err) => {
        this.destroy(err)
      })
      .on('data', (chunk) => {
        this._currentLength += chunk.length
        if (this._totalLength) {
          this.emit('progress', Math.round(this._currentLength / this._totalLength * 100))
        }
        this.push(chunk)
      })
  }

  _startFetch() {
    fetch(this._url)
      .then((res) => {
        if (!res.ok) {
          return this.destroy(new Error(`fetch resulted in ${res.status}`))
        }
        this._totalLength = res.headers.get('content-length')
        this._processData(res.body)
      })
      .catch((err) => {
        return this.destroy(new Error(err))
      })
  }

  _read() {
    if (!this._fetchStarted) {
      this._fetchStarted = true
      this._startFetch()
    }
  }
}

module.exports.loadFile = (url) => new FileLoader(url)

And the consumer code:

// consumer.js
const fs = require('fs')
const {loadFile} = require('./fileLoader')

loadFile('http://example.com/video.mp4')
  .on('progress', (progress) => {
    console.log(`${progress}%`)
  })
  .on('end', () => {
    console.log('done')
  })
  .on('error', (err) => {
    console.log(err)
  })
  .pipe(fs.createWriteStream('./tempy.mp4'))

Upvotes: 2

Related Questions