more resilient timestamp handling

lib/reader.js

@@ -128,13 +128,8 @@ function HlsStreamReader(src, options) {
           if (index.variant)
             return self.end();
 
-          if (!self.readState.currentSegment) {
-            self.readState.currentSegment = index.getSegment(self.readState.currentSeq);
-            if (self.readState.currentSegment)
-              fetchfrom(self.readState.currentSegment);
-            else if (index.ended)
-              self.end();
-          }
+          if (!self.readState.currentSegment)
+            checkcurrent();
         }
 
         if (!index.ended) {
@@ -147,6 +142,19 @@ function HlsStreamReader(src, options) {
   }
   updateindex();
 
+  function checkcurrent() {
+    self.readState.currentSegment = self.index.getSegment(self.readState.currentSeq);
+    if (self.readState.currentSegment)
+      fetchfrom(self.readState.currentSegment);
+    else if (self.index.ended)
+      self.end();
+    else if (!self.index.type && (self.index.lastSeqNo() < self.readState.currentSeq-1)) {
+      // handle live stream restart
+      self.readState.currentSeq = self.index.first_seq_no;
+      checkcurrent();
+    }
+  }
+
   function fetchfrom(segment) {
     var segmentUrl = url.resolve(self.baseUrl, segment.uri)
 
@@ -163,11 +171,7 @@ function HlsStreamReader(src, options) {
 
       function nextstream() {
         self.readState.currentSeq++;
-        self.readState.currentSegment = self.index.getSegment(self.readState.currentSeq);
-        if (self.readState.currentSegment)
-          fetchfrom(self.readState.currentSegment);
-        else if (self.index.ended)
-          self.end();
+        checkcurrent();
       }
 
       if (stream) {

lib/tslimit.js

@@ -52,7 +52,13 @@ function TsLimit() {
     var pcr_delta = pcr - self.pcr;
     if (pcr_delta < 0) pcr_delta += (0x200000000 * 300) / 27;
 
-    return self.last + pcr_delta;
+    var ret = self.last + pcr_delta;
+    if (pcr_delta > 3600E6) {
+      // update pcr reference every hour to handle wrap-around
+      self.pcr = pcr;
+      self.last = ret;
+    }
+    return ret;
   }
 
   this.bytes2delta = function(bytes) {
@@ -82,8 +88,15 @@ TsLimit.prototype._transform = function(chunk, output, cb) {
     for (var i=0; i<end; i+=188) {
       self.time += packet_time;
       var pcr = parsePCR(buf.slice(i, i+188));
-      if (pcr !== -1)
-        self.time = self.pcr2time(pcr);
+      if (pcr !== -1) {
+        var pcrtime = self.pcr2time(pcr);
+        if (Math.abs(pcrtime - self.time) > 700E3) {
+          console.error('PCR out of sync');
+          self.pcr = -1;
+          pcrtime = self.pcr2time(pcr);
+        }
+        self.time = pcrtime;
+      }
       if (self.time > now)
         break;
     }
@@ -93,7 +106,7 @@ TsLimit.prototype._transform = function(chunk, output, cb) {
     self.buffer = buf.slice(i);
 
     if (i < end) {
-      return setTimeout(process, (self.time - now) / 1000);
+      return setTimeout(process, Math.min(50000, (self.time - now) / 1000));
     }
     cb();
   }