reset after rate errors

this causes better recovery from prolonged missing / slow connection.

bin/hlsdump

@@ -98,28 +98,53 @@ r.on('end', function() {
 });
 
 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;
 
-  if (hlsdump.sync)
-    stream = stream.pipe(tssmooth());
+  console.error('hooking output');
+
+  var s = stream;
+  if (hlsdump.sync) {
+    var smooth = tssmooth();
+    smooth.on('unpipe', function() {
+      this.unpipe();
+    });
+    smooth.once('error', function(err) {
+      console.error('smooth error', err);
+      unhook(stream);
+
+      smooth.on('error', function () {});
+    });
+    s = s.pipe(smooth);
+  }
 
-  if (hlsdump.udp)
-    stream.pipe(tsblast(hlsdump.udp));
+  outputs.forEach(function (o) {
+    s.pipe(o);
+  });
 
-  if (hlsdump.output) {
-    var dst;
-    if (hlsdump.output === '-')
-      dst = process.stdout;
-    else
-      dst = fs.createWriteStream(hlsdump.output);
+  hooked = true;
+}
 
-    if (dst)
-      stream.pipe(dst);
+function unhook(stream) {
+  if (hooked) {
+    console.error('unhooking output');
+    stream.unpipe();
+    hooked = false;
   }
-
-  hooked = true;
 }
 
 if (!hlsdump.sync)

lib/tssmooth.js

@@ -14,6 +14,14 @@ try {
 module.exports = tssmooth;
 exports.TsSmooth = TsSmooth;
 
+function RateError(msg) {
+  Error.call(this);
+
+  this.message = msg;
+}
+util.inherits(RateError, Error);
+RateError.prototype.name = 'Rate Error';
+
 function parsePCR(buffer, index, pcr_pid) {
   var head = buffer.readUInt32BE(index, true);
   var pid = (head >> 8) & 0x1fff;
@@ -83,15 +91,18 @@ function TsSmooth(options) {
     var delta = pcrtime - self.pcrtime;
     if (delta > 100E3 || delta < 0) {
       console.error('PCR_error: '+(delta/1E6).toFixed(2)+'s missing');
-      var now = utime();
-      var error = now - pcrtime;
-      if (Math.abs(error) > 60*1E6) {
-        console.error('PCR sync reset from '+(error/1E6).toFixed(2)+'s error');
-        self.pcr = -1;
-        pcrtime = self.pcr2time(newPCR);
-      }
     }
+
+    var now = utime();
+    var error = now - pcrtime;
+    if (Math.abs(error) > 20*1E6) {
+      self.emit('error', new RateError('PCR sync offset '+(error/1E6).toFixed(2)+'s error'));
+      self.pcr = -1;
+      pcrtime = self.pcr2time(newPCR);
+    }
+
     self.pcrtime = pcrtime;
+
     return pcrtime;
   }
 
@@ -150,7 +161,7 @@ TsSmooth.prototype._transform = function(chunk, encoding, cb) {
             break;
         }
         // remove bad data
-        console.error('slice', sync, end);
+        debug('slice', sync, end);
         buf = Buffer.concat([buf.slice(0, index), buf.slice(sync)]);
         end -= sync-index;
         continue;