don't try to smooth over more than 80 ms

lib/tssmooth.js

@@ -80,42 +80,44 @@ function TsSmooth(options) {
   }
 
   this.output_time = function(newPCR) {
-    if (newPCR === -1) return -1;
+    // when this is called normally, now ~= self.pcrtime
+    if (newPCR === -1) return;
 
     var pcrtime = self.pcr2time(newPCR);
     if (self.pcrtime === -1) {
       self.pcrtime = pcrtime;
-      return -1;
+      return;
     }
 
     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) > 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;
+    return { time:pcrtime, delta: delta };
   }
 
   Transform.call(this, {highWaterMark:this.packetSize});
 }
 util.inherits(TsSmooth, Transform);
 
+TsSmooth.prototype.reset = function(currentPCR) {
+  this.pcr = -1;
+  if (typeof currentPCR !== 'undefined')
+    this.pcrtime = this.pcr2time(currentPCR);
+};
+
 function utime() {
   var t = process.hrtime(); // based on CLOCK_MONOTONIC, and thus accommodates local drift (but apparently not suspend)
 //  console.error(t);
   return t[0] * 1E6 + t[1] / 1E3;
 }
 
+function wait(waitMs, fn) {
+  if (waitMs > 0)
+    setTimeout(fn, waitMs);
+  else
+    fn();
+}
+
 // smoothly outputs given buffer before endTime
 function outputBefore(stream, buffer, endTime, packetSize, cb) {
   var index = 0;
@@ -166,11 +168,29 @@ TsSmooth.prototype._transform = function(chunk, encoding, cb) {
       }
 
       var pcr = parsePCR(buf, index);
-      var outtime = self.output_time(pcr);
-      if (outtime !== -1 && index !== startIndex) {
-        var slice = buf.slice(startIndex, index);
-        startIndex = index;
-        return outputBefore(self, slice, outtime, self.packetSize, processNext);
+      var out = self.output_time(pcr);
+      if (out !== undefined && index !== startIndex) {
+        if (out.delta > 100E3 || out.delta < 0)
+          self.emit('warning', new Error('PCR_error: '+(out.delta/1E6).toFixed(2)+'s missing'));
+
+        var now = utime();
+        var error = (out.time - now) - out.delta;
+        var waittime = (error > 80000) ? (error/1000 - 5) : 0;
+
+        debug('error', (error/1E6).toFixed(2));
+        if (error < -2*1E6 || error > 300*1E6) {
+          // negative == buffer too late
+          // positive == buffer too early
+          self.emit('warning', new RateError('PCR sync offset '+(error/1E6).toFixed(2)+'s error'));
+          self.reset(pcr);
+          waittime = 0;
+        }
+
+        return wait(waittime, function output() {
+          var slice = buf.slice(startIndex, index);
+          startIndex = index;
+          return outputBefore(self, slice, out.time, self.packetSize, processNext);
+        });
       }
       index += 188;
     }