node-hls-tools/bin/hlsdump

#!/usr/bin/env node

"use strict";

var hlsdump = require('commander');
hlsdump.version('0.0.0')
   .usage('[options] <url>')
   .option('-o, --output <path>', 'target file')
   .option('-u, --udp [host:port]', 'relay TS over UDP', function(val) {
     var r = { host:'localhost', port:1234 };
     if (val) {
       var s = val.split(':');
       if (s.length === 1) {
         r.port = parseInt(s[0], 10);
       } else {
         r.host = s[0];
         r.port = parseInt(s[1], 10);
       }
     }
     return r;
   })
   .option('-b, --buffer-size <bytes>|full', 'try to buffer <bytes> of input data (implies -s)', function(val) {
     if (val === 'full') return 0x80000000-1;
     return parseInt(val, 0);
   })
   .option('-s, --sync', 'clock sync using stream PCR')
   .option('-f, --full-stream', 'fetch all stream data')
   .option('-c, --concurrent <count>', 'fetch using concurrent connections', parseInt)
   .option('-a, --user-agent <string>', 'HTTP User-Agent')
   .parse(process.argv);

var util = require('util'),
    url = require('url'),
    fs = require('fs');

var reader = require('../lib/reader'),
    tssmooth = require('../lib/tssmooth'),
    tsblast = require('../lib/tsblast'),
    oncemore = require('../lib/oncemore');

try {
  var Passthrough = require('stream').Passthrough;
  assert(Passthrough);
} catch (e) {
  var Passthrough = require('readable-stream/passthrough');
}

var src = hlsdump.args[0];
if (!src) return hlsdump.help();

if (hlsdump.bufferSize) hlsdump.sync = true;

var r = reader(src, {highWaterMark:(hlsdump.concurrent || 1) - 1, fullStream:hlsdump.fullStream});

var time = 0;
var reading = false;
r.on('readable', function() {
  if (reading) return;// console.error('readable call error');

  function grabnext() {
    var obj = r.read();
    if (obj) {
      var meta = obj.meta;
      var duration = obj.segment.duration;
      var size = meta.size;
      time += duration;

      reading = true;
      obj.stream.pipe(buffer, { end: false });
      if (size === -1 || !hooked) {
        size = 0;
        obj.stream.on('data', function(chunk) {
          size += chunk.length;
          if (!hooked && size >= hlsdump.bufferSize)
            hook(buffer);
        });
      }
      oncemore(obj.stream).once('end', 'error', function(err) {
        reading = false;
        console.error('segment done at '+time.toFixed(0)+' seconds, avg bitrate (kbps):', (size / (duration * 1024/8)).toFixed(1));
        if (err) console.error('stream error', err.stack || err);
        hook(buffer);
        grabnext();
      });
    }
  }
  grabnext();
});

r.once('index', function() {
  // wait until first index is returned before attaching error listener.
  // this will enable initials errors to throw
  r.on('error', function(err) {
    console.error('reader error', err.stack || err);
  });
});

r.on('end', function() {
  console.error('done');
});

var buffer = new Passthrough({highWaterMark:hlsdump.bufferSize});
var outputs = [];

if (hlsdump.udp)
  outputs.push(tsblast(hlsdump.udp));

if (hlsdump.output) {
  if (hlsdump.output === '-')
    outputs.push(process.stdout);
  else
    outputs.push(fs.createWriteStream(hlsdump.output));
}

// the hook is used to prebuffer
var hooked = false;
function hook(stream) {
  if (hooked) return;

  console.error('hooking output');

  var s = stream;
  if (hlsdump.sync) {
    var smooth = tssmooth();
    smooth.on('unpipe', function() {
      this.unpipe();
    });
    smooth.on('error', function(err) {
      console.error('smooth error', err);
    });
    s = s.pipe(smooth);
  }

  outputs.forEach(function (o) {
    s.pipe(o);
  });

  hooked = true;
}

if (!hlsdump.sync)
  hook(buffer);