Merge pull request #11 from videojs/feature/async-appends

Yield control to the browser between appending FLV tags

@@ -22,7 +22,9 @@ module.exports = function(grunt) {
         stripBanners: true
       },
       dist: {
-        src: ['src/video-js-hls.js',
+        nonull: true,
+        src: ['src/videojs-hls.js',
+              'src/async-queue.js',
               'src/flv-tag.js',
               'src/exp-golomb.js',
               'src/h264-stream.js',

@@ -14,6 +14,7 @@
 
   <!-- HLS plugin -->
   <script src="src/videojs-hls.js"></script>
+  <script src="src/async-queue.js"></script>
 
   <!-- segment handling -->
   <script src="src/flv-tag.js"></script>

+(function(window, videojs, undefined) {
+  'use strict';
+  /**
+   * A queue object that manages tasks that should be processed
+   * serially but asynchronously. Loosely adapted from
+   * https://github.com/caolan/async#queue.
+   * @param worker {function} the callback to invoke with each value
+   * pushed onto the queue
+   * @return {object} an object with an array of `tasks` that remain to
+   * be processed and function `push` to add new tasks
+   */
+  videojs.hls.queue = function(worker) {
+    var
+      q = {
+        tasks: [],
+        running: false,
+        push: function(task) {
+          q.tasks.push(task);
+          if (!q.running) {
+            window.setTimeout(process, 0);
+            q.running = true;
+          }
+        },
+      },
+      process = function() {
+        var task;
+        if (q.tasks.length) {
+          task = q.tasks.shift();
+          worker.call(this, task);
+          window.setTimeout(process, 0);
+        } else {
+          q.running = false;
+        }
+      };
+    return q;
+  };
+})(window, window.videojs);

@@ -187,6 +187,15 @@ var
       mediaSource = new videojs.MediaSource(),
       segmentParser = new videojs.hls.SegmentParser(),
       player = this,
+
+      // async queue of Uint8Arrays to be appended to the SourceBuffer
+      tags = videojs.hls.queue(function(tag) {
+        player.hls.sourceBuffer.appendBuffer(tag, player);
+
+        if (player.hls.mediaIndex === player.hls.media.segments.length) {
+          mediaSource.endOfStream();
+        }
+      }),
       srcUrl,
 
       segmentXhr,
@@ -268,9 +277,14 @@ var
     player.on('seeking', function() {
       var currentTime = player.currentTime();
       player.hls.mediaIndex = getMediaIndexByTime(player.hls.media, currentTime);
+
+      // cancel outstanding requests and buffer appends
       if (segmentXhr) {
         segmentXhr.abort();
       }
+      tags.tasks = [];
+
+      // begin filling the buffer at the new position
       fillBuffer(currentTime * 1000);
     });
 
@@ -535,16 +549,14 @@ var
         }
 
         while (segmentParser.tagsAvailable()) {
-          player.hls.sourceBuffer.appendBuffer(segmentParser.getNextTag().bytes, player);
+          // queue up the bytes to be appended to the SourceBuffer
+          // the queue gives control back to the browser between tags
+          // so that large segments don't cause a "hiccup" in playback
+          tags.push(segmentParser.getNextTag().bytes);
         }
 
         player.hls.mediaIndex++;
 
-        if (player.hls.mediaIndex === player.hls.media.segments.length) {
-          mediaSource.endOfStream();
-          return;
-        }
-
         // figure out what stream the next segment should be downloaded from
         // with the updated bandwidth information
         playlist = player.hls.selectPlaylist();

+(function(window, queue, undefined) {
+  var
+    oldSetTimeout,
+    callbacks;
+  module('async queue', {
+    setup: function() {
+      oldSetTimeout = window.setTimeout;
+      callbacks = [];
+      window.setTimeout = function(callback) {
+        callbacks.push(callback);
+      };
+    },
+    teardown: function() {
+      window.setTimeout = oldSetTimeout;
+    }
+  });
+
+  test('runs tasks asynchronously', function() {
+    var
+      run = false,
+      q = queue(function() {
+        run = true;
+      });
+    q.push(1);
+    
+    ok(!run, 'tasks are not run immediately');
+
+    callbacks[0]();
+    ok(run, 'tasks are run asynchronously');
+  });
+
+  test('runs one task at a time', function() {
+    var q = queue(function() {});
+    q.push(1);
+    q.push(2);
+    q.push(3);
+    q.push(4);
+    q.push(5);
+
+    strictEqual(q.tasks.length, 5, 'all tasks are queued');
+    strictEqual(1, callbacks.length, 'one callback is registered');
+  });
+
+  test('tasks are scheduled until the queue is empty', function() {
+    var q = queue(function() {});
+    q.push(1);
+    q.push(2);
+
+    callbacks.shift()();
+    strictEqual(1, callbacks.length, 'the next task is scheduled');
+
+    callbacks.shift()();
+    strictEqual(1, callbacks.length, 'nothing is scheduled on an empty queue');
+  });
+
+  test('can be emptied at any time', function() {
+    var
+      runs = 0,
+      q = queue(function() {
+        runs++;
+      });
+    q.push(1);
+    q.push(2);
+
+    callbacks.shift()();
+    strictEqual(1, runs, 'task one is run');
+
+    q.tasks = [];
+    callbacks.shift()();
+    strictEqual(1, runs, 'the remaining tasks are cancelled');
+  });
+})(window, window.videojs.hls.queue);

@@ -31,6 +31,9 @@
   <script src="tsSegment-bc.js"></script>
   <script src="../src/bin-utils.js"></script>
 
+  <!-- async queue -->
+  <script src="../src/async-queue.js"></script>
+
   <!-- Test cases -->
   <script>
   module('environment');
@@ -44,6 +47,7 @@
   <script src="exp-golomb_test.js"></script>
   <script src="flv-tag_test.js"></script>
   <script src="m3u8_test.js"></script>
+  <script src="async-queue_test.js"></script>
 </head>
 <body>
   <div id="qunit"></div>

@@ -25,9 +25,32 @@ var
   oldFlashSupported,
   oldXhr,
   oldSegmentParser,
+  oldSetTimeout,
   oldSourceBuffer,
   oldSupportsNativeHls,
-  xhrUrls;
+  xhrUrls,
+
+  mockSegmentParser = function(tags) {
+    if (tags === undefined) {
+      tags = [];
+    }
+    return function() {
+      this.getFlvHeader = function() {
+        return 'flv';
+      };
+      this.parseSegmentBinaryData = function() {};
+      this.flushTags = function() {};
+      this.tagsAvailable = function() {
+        return tags.length;
+      };
+      this.getTags = function() {
+        return tags;
+      };
+      this.getNextTag = function() {
+        return tags.shift();
+      };
+    };
+  };
 
 module('HLS', {
   setup: function() {
@@ -59,10 +82,11 @@ module('HLS', {
       return videojs.createTimeRange(0, 0);
     };
 
-    // store the SegmentParser so it can be easily mocked
+    // store functionality that some tests need to mock
     oldSegmentParser = videojs.hls.SegmentParser;
+    oldSetTimeout = window.setTimeout;
 
-    // make XHR synchronous
+    // make XHRs synchronous
     oldXhr = window.XMLHttpRequest;
     window.XMLHttpRequest = function() {
       this.open = function(method, url) {
@@ -89,6 +113,7 @@ module('HLS', {
     videojs.hls.supportsNativeHls = oldSupportsNativeHls;
     videojs.hls.SegmentParser = oldSegmentParser;
     videojs.SourceBuffer = oldSourceBuffer;
+    window.setTimeout = oldSetTimeout;
     window.XMLHttpRequest = oldXhr;
   }
 });
@@ -665,53 +690,88 @@ test('flushes the parser after each segment', function() {
 test('drops tags before the target timestamp when seeking', function() {
   var
     i = 10,
-    segment = [],
+    callbacks = [],
     tags = [],
     bytes = [];
 
   // mock out the parser and source buffer
-  videojs.hls.SegmentParser = function() {
-    this.getFlvHeader = function() {
-      return [];
-    };
-    this.parseSegmentBinaryData = function() {};
-    this.flushTags = function() {};
-    this.tagsAvailable = function() {
-      return tags.length;
-    };
-    this.getTags = function() {
-      return tags;
-    };
-    this.getNextTag = function() {
-      return tags.shift();
-    };
-  };
+  videojs.hls.SegmentParser = mockSegmentParser(tags);
   window.videojs.SourceBuffer = function() {
     this.appendBuffer = function(chunk) {
       bytes.push(chunk);
     };
   };
+  // capture timeouts
+  window.setTimeout = function(callback) {
+    callbacks.push(callback);
+  };
+
+  // push a tag into the buffer
+  tags.push({ pts: 0, bytes: 0 });
 
   player.hls('manifest/media.m3u8');
   videojs.mediaSources[player.currentSrc()].trigger({
     type: 'sourceopen'
   });
+  while (callbacks.length) {
+    callbacks.shift()();
+  }
 
-  // build up some mock FLV tags
+  // mock out a new segment of FLV tags
+  bytes = [];
   while (i--) {
-    segment.unshift({
+    tags.unshift({
       pts: i * 1000,
       bytes: i
     });
   }
-  tags = segment.slice();
-  bytes = [];
   player.currentTime = function() {
     return 7;
   };
   player.trigger('seeking');
 
-  deepEqual([7,8,9], bytes, 'three tags are appended');
+  while (callbacks.length) {
+    callbacks.shift()();
+  }
+
+  deepEqual(bytes, [7,8,9], 'three tags are appended');
+});
+
+test('clears pending buffer updates when seeking', function() {
+  var
+    bytes = [],
+    callbacks = [],
+    tags = [{ pts: 0, bytes: 0 }];
+  // mock out the parser and source buffer
+  videojs.hls.SegmentParser = mockSegmentParser(tags);
+  window.videojs.SourceBuffer = function() {
+    this.appendBuffer = function(chunk) {
+      bytes.push(chunk);
+    };
+  };
+  // capture timeouts
+  window.setTimeout = function(callback) {
+    callbacks.push(callback);
+  };
+
+  // queue up a tag to be pushed into the buffer (but don't push it yet!)
+  player.hls('manifest/media.m3u8');
+  videojs.mediaSources[player.currentSrc()].trigger({
+    type: 'sourceopen'
+  });
+
+  // seek to 7s
+  tags.push({ pts: 7000, bytes: 7 });
+  player.currentTime = function() {
+    return 7;
+  };
+  player.trigger('seeking');
+
+  while (callbacks.length) {
+    callbacks.shift()();
+  }
+
+  deepEqual(bytes, ['flv', 7], 'tags queued to be appended should be cancelled');
 });
 
 test('playlist 404 should trigger MEDIA_ERR_NETWORK', function() {