refactor(): initial Source+Delivery class refactor

.github/workflows/nodejs.yml

@@ -0,0 +1,25 @@
+name: Node CI
+
+on: [push]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        node-version: [12.x]
+
+    steps:
+    - uses: actions/checkout@v1
+    - name: Use Node.js ${{ matrix.node-version }}
+      uses: actions/setup-node@v1
+      with:
+        node-version: ${{ matrix.node-version }}
+    - name: npm install, build, and test
+      run: |
+        npm i
+        npm test
+      env:
+        CI: true

.gitignore

@@ -60,3 +60,4 @@ typings/
 # next.js build output
 .next
 
+package-lock.json

.npmrc

@@ -0,0 +1,2 @@
+tag-version-prefix=""
+message="chore(release): %s :tada:"

asset.js

@@ -1,7 +1,7 @@
 const path = require('path')
-const { Delivery, Package, Source, Track } = require('./index')
 const Target = require('./target')
 const uuidv4 = require('uuid/v4')
+const { Delivery, Package, Source, Track } = require('./index')
 
 class Asset {
   constructor(options = {}) {

delivery.js

@@ -0,0 +1,171 @@
+const { EventEmitter } = require('events')
+const { Source } = require('./source')
+const { Pool } = require('nanoresource-pool')
+const assert = require('assert')
+const Batch = require('batch')
+const uuid = require('uuid/v4')
+const os = require('os')
+
+/**
+ * Default concurrency of source probing (`ffprobe`).
+ * @private
+ */
+const DEFAULT_PROBE_CONCURRENCY = 2 * os.cpus().length
+
+/**
+ * Default concurrency of source probing (`ffmpeg`).
+ * @private
+ */
+const DEFAULT_DEMUX_CONCURRENCY = os.cpus().length
+
+/**
+ * The `Delivery` class represents a container of multiple
+ * sources.
+ * @public
+ * @class
+ * @extends nanoresource-pool
+ */
+class Delivery extends Pool {
+
+  /**
+   * `Delivery` class constructor.
+   * @param {?(Object)} opts
+   * @param {?(String)} opts.id
+   */
+  constructor(opts) {
+    super(Source, opts)
+
+    if (!opts || 'object' !== typeof opts) {
+      opts = {}
+    }
+
+    this.id = opts.id || uuid()
+  }
+
+  /**
+   * All sources in the delivery pool, including
+   * child delivery pools.
+   * @accessor
+   * @type {Array<Source>}
+   */
+  get sources() {
+    return this.query()
+  }
+
+  /**
+   * Creates and adds a new source from a URI.
+   * @param {String} uri
+   * @param {?(Object} opts
+   * @return {Source}
+   */
+  source(uri, opts) {
+    return this.resource(uri, opts)
+  }
+
+  /**
+   * Probes all sources in delivery pool.
+   * @param {?(Object} opts
+   * @param {Function} callback
+   */
+  probe(opts, callback) {
+    if ('function' === typeof opts) {
+      callback = opts
+    }
+
+    if (!opts || 'object' !== typeof opts) {
+      opts = {}
+    }
+
+    assert('function' === typeof callback, 'callback is not a function')
+
+    const { concurrency = DEFAULT_PROBE_CONCURRENCY } = opts
+    const { sources } = this
+    const probes = {}
+    const batch = new Batch().concurrency(concurrency)
+
+    for (const source of sources) {
+      batch.push((next) => {
+        source.probe((err, info) => {
+          if (err) { return next(err) }
+          probes[source.uri] = info
+          next(null)
+        })
+      })
+    }
+
+    batch.end((err) => {
+      if (err) { return callback(err) }
+      callback(null, probes)
+    })
+  }
+
+  /**
+   * An alias for `probe()`.
+   * @param {Function} callback
+   */
+  stat(callback) {
+    this.probe(callback)
+  }
+
+  /**
+   * Demux sources into output streams. Outputs
+   * @param {?(Object)} opts
+   * @param {Function} callback
+   */
+  demux(opts, callback) {
+    if ('function' === typeof opts) {
+      callback = opts
+    }
+
+    if (!opts || 'object' !== typeof opts) {
+      opts = {}
+    }
+
+    assert('function' === typeof callback, 'callback is not a function')
+
+    const { concurrency = DEFAULT_PROBE_CONCURRENCY } = opts
+    const { sources } = this
+    const demuxes = {}
+    const emitter = new EventEmitter()
+    const batch = new Batch().concurrency(concurrency)
+
+    emitter.setMaxListeners(0)
+
+    for (const source of sources) {
+      batch.push((next) => {
+        const demux = source.demux(opts, (err, outputs) => {
+          if (err) { return next(err) }
+          demuxes[source.uri] = outputs
+          next(null)
+        })
+
+        const proxy = (event) => {
+          demux.on(event, (...args) => {
+            emitter.emit(event, ...args.concat(source))
+          })
+        }
+
+        proxy('codecData')
+        proxy('end')
+        proxy('error')
+        proxy('progress')
+        proxy('start')
+        proxy('stderr')
+      })
+    }
+
+    batch.end((err) => {
+      if (err) { return callback(err) }
+      callback(null, demuxes)
+    })
+
+    return emitter
+  }
+}
+
+/**
+ * Module exports.
+ */
+module.exports = {
+  Delivery
+}

example/basic-delivery/index.js

@@ -0,0 +1,12 @@
+const { Delivery } = require('../../delivery')
+const { Source } = require('../../source')
+
+const uri = 'http://distribution.bbb3d.renderfarming.net/video/mp4/bbb_sunflower_1080p_60fps_normal.mp4'
+
+const delivery = new Delivery()
+
+delivery.ready(() => {
+  delivery.source(uri)
+  delivery.probe(console.log)
+  //delivery.demux(console.log).on('progress', console.log)
+})

example/source-demux/index.js

@@ -0,0 +1,86 @@
+const prettyBytes = require('pretty-bytes')
+const { Source } = require('../../source')
+const prettyTime = require('pretty-ms')
+const Timecode = require('smpte-timecode')
+const Progress = require('progress')
+const path = require('path')
+
+const uri = 'http://distribution.bbb3d.renderfarming.net/video/mp4/bbb_sunflower_1080p_60fps_normal.mp4'
+const source = new Source(uri)
+
+console.log('>  open: %s', uri)
+source.open((err) => {
+  console.log('>  open: %s %s (ready)',
+    prettyBytes(source.byteLength),
+    path.basename(source.uri))
+
+
+  console.log('> probe: %s', path.basename(source.uri))
+  source.probe((err, probe) => {
+    //console.log(probe);
+    for (let i = 0; i < probe.streams.length; ++i) {
+      console.log(`> probe: streams[${i}]: %s codec_type=%s bit_rate=%s nb_frames=%s`,
+        probe.streams[i].codec_long_name,
+        probe.streams[i].codec_type,
+        probe.streams[i].bit_rate,
+        probe.streams[i].nb_frames,
+      )
+
+      if (probe.streams[i].width) {
+        console.log(`> probe: streams[${i}]: width=%s height=%s`,
+          probe.streams[i].width,
+          probe.streams[i].height,
+        )
+      }
+
+      if (probe.streams[i].pix_fmt) {
+        console.log(`> probe: streams[${i}]: pix_fmt=%s`,
+          probe.streams[i].pix_fmt,
+        )
+      }
+    }
+
+    console.log('> probe: format: %s', probe.format.format_long_name)
+    console.log('> probe: format: %s', probe.format.tags.comment)
+    console.log('> probe: format: %s - %s',
+      probe.format.tags.title,
+      probe.format.tags.artist)
+
+    const progress = new Progress(`> demux: | :timecode | (:frames/:nb_frames frames) [:bar] :percent`, {
+      width: 32,
+      total: probe.streams[0].nb_frames
+    })
+
+    let timecode = null
+    const demux = source.demux((err, outputs) => {
+      if (err) {
+        console.error('error:', err.message)
+      } else {
+        console.log('> demux: outputs:')
+        console.log(outputs
+          .map((output) => output.replace(process.cwd(), '.'))
+          .join('\n'))
+      }
+    })
+
+    demux.on('progress', (info) => {
+      try {
+        const dropFrameSupport = [29.97, 59.94]
+        const [num, den] = probe.streams[0].r_frame_rate.split('/')
+
+        const t = new Date(1970, 0, 1)
+        t.setSeconds(probe.streams[0].duration)
+
+        const fps = Number((num/den).toFixed(3))
+        timecode = new Timecode(info.timemark, fps, (num % den !== 0 && dropFrameSupport.includes(fps)))
+      } catch (err) {
+      }
+
+      progress.update(info.frames / probe.streams[0].nb_frames, {
+        timecode: timecode ? String(timecode) : info.timemark,
+        frames: info.frames,
+        nb_frames: probe.streams[0].nb_frames
+      })
+    })
+  })
+})

example/source-probe/index.js

@@ -0,0 +1,12 @@
+const prettyBytes = require('pretty-bytes')
+const { Source } = require('../../source')
+const path = require('path')
+
+const uri = 'http://distribution.bbb3d.renderfarming.net/video/mp4/bbb_sunflower_1080p_60fps_normal.mp4'
+const source = new Source(uri)
+source.open((err) => {
+  console.log('%s %s (ready)',
+    prettyBytes(source.byteLength),
+    path.basename(source.uri))
+  source.probe(console.log)
+})

ffmpeg.js

@@ -0,0 +1,26 @@
+const ffmpeg = require('fluent-ffmpeg')
+
+/**
+ * Static `ffmpeg(1)` binary path.
+ * @public
+ */
+const FFPROBE_BIN_PATH = require('ffprobe-static').path
+
+/**
+ * Static `ffprobe(1)` binary path.
+ * @public
+ */
+const FFMPEG_BIN_PATH = require('ffmpeg-static').path
+
+// configure ffmpeg/ffprobe
+ffmpeg.setFfmpegPath(FFMPEG_BIN_PATH)
+ffmpeg.setFfprobePath(FFPROBE_BIN_PATH)
+
+/**
+ * Module exports.
+ */
+module.exports = {
+  FFPROBE_BIN_PATH,
+  FFMPEG_BIN_PATH,
+  ffmpeg
+}

mkvmerge.js

@@ -0,0 +1,19 @@
+const { platform } = process
+
+/**
+ * Static `mkvmerge(1)` binary path. Will be `null`
+ * if platform is not supported.
+ * @public
+ */
+const MKVMERGE_BIN_PATH = 'linux' === platform
+  ? require('mkvmerge-static-linux').path
+  : ['win32', 'darwin'].includes(platform)
+    ? require('mkvmerge-static').path
+    : null
+
+/**
+ * Module.exports
+ */
+module.exports = {
+  MKVMERGE_BIN_PATH
+}