Segmenting videos (HLS support)

Added `segments_options` to job input options, which only can contain
a `segment_time` time parameter so far.

Also added `playlist` and `segments_options` columns to the Job model and API
ouput.

This feature adds as a "post-processing" action and requires
`encoder_options` to be set. They also should prepare file to be
"segmentable". Not all files can be segmented without re-encoding.

Segmenting applies on ready `destination_file`.
It generates playlist named "destination_file_basename.m3u8" and a set
of segments in a mpegts format named "destination_file_basename-%06d.ts"

Also method `Job#finalize` refactored a bit. Moved few guard clauses to
the top instead of 3-levels if-else conditions.

Related to #31

Author: cutalion

lib/job-handler.js

@@ -98,14 +98,15 @@ function validateJobRequest(postData, callback) {
   // destination_file:  the output file
   // encoder_options:   the flags for the encoder
   // thumbnail_options: options to generate thumbnails (optional)
+  // segments_options:  options to generate segmented video (optional)
   // callback_urls:     array of callbacks to notify of events (optional)
   var missingFields = [];
   var opts = {};
 
   try {
     var obj = JSON.parse(postData);
     var requiredFields = ['source_file', 'destination_file', 'encoder_options'];
-    var acceptedFields = ['source_file', 'destination_file', 'encoder_options', 'thumbnail_options', 'callback_urls'];
+    var acceptedFields = ['source_file', 'destination_file', 'encoder_options', 'thumbnail_options', 'segments_options', 'callback_urls'];
 
     for (var field in requiredFields) {
       if (typeof(obj[requiredFields[field]]) == "undefined") {
@@ -157,4 +158,4 @@ function hasFreeSlots() {
 function removeItemFromSlot(item) {
   var idx = slots.indexOf(item);
   if(idx!=-1) slots.splice(idx, 1);
-}
+}

lib/job.js

@@ -144,6 +144,8 @@ var Job = JobUtils.getDatabase().define('Job', {
   opts:        { type: Sequelize.TEXT, defaultValue: null },
   thumbnails:  { type: Sequelize.TEXT, defaultValue: null },
   message:     { type: Sequelize.TEXT, defaultValue: null },
+  playlist:    { type: Sequelize.STRING, defaultValue: null },
+  segments:    { type: Sequelize.TEXT, defaultValue: null },
   createdAt:   Sequelize.DATE,
   updatedAt:   Sequelize.DATE
 }, {
@@ -274,35 +276,103 @@ var Job = JobUtils.getDatabase().define('Job', {
       }
     },
     didFinish: function(code) {
-      if (code == 0 && this.parsedOpts()['thumbnail_options']) {
-        this.processThumbnails();
-      } else {
+      if (code != 0) {
         this.finalize(code);
+        return;
       }
+
+      this.processThumbnails({
+        error:   function(job) { job.finalize(1); },
+        success: function(job) {
+          job.processSegments({
+            error:   function(job) { job.finalize(1); },
+            success: function(job) { job.finalize(0); }
+          })
+        }
+      })
     },
-    processThumbnails: function() {
+
+    processSegments: function(callbacks){
+      if (!this.parsedOpts()['segments_options']) {
+        callbacks.success(this);
+        return;
+      }
+
+      logger.log("Processing segments for job " + this.internalId + ".");
+
+      var job              = this;
+      var args             = [];
+      var segmentsOpts     = this.parsedOpts()['segments_options'];
+      var segmentTime      = segmentsOpts['segment_time'];
+      var destinationFile  = this.parsedOpts()['destination_file'];
+      var playlistName     = path.basename(destinationFile, path.extname(destinationFile));
+      var playlistDir      = path.dirname(destinationFile);
+      var playlistPath     = [playlistDir, playlistName].join(path.sep) + '.m3u8';
+      var segmentsFormat   = [playlistDir, playlistName].join(path.sep) + '-%06d.ts'
+
+      args.push('-i', job.tmpFile,
+                '-codec', 'copy', '-map', '0', '-f', 'segment',
+                '-vbsf', 'h264_mp4toannexb', '-flags', '-global_header',
+                '-segment_format', 'mpegts', '-segment_list', playlistPath,
+                '-segment_time', segmentTime, segmentsFormat);
+
+      child_process.execFile(config['encoder'], args, function(error, stdout, stderr) {
+        if (error) {
+          job.lastMessage = 'Error while generating segments: ' + error.message;
+          callbacks.error(job);
+          return;
+        }
+
+        fs.readdir(playlistDir, function(error, files) {
+          if (error) {
+            job.lastMessage = 'Error while generating segments: ' + error.message;
+            callbacks.error(job);
+            return;
+          }
+
+          job.segments = JSON.stringify(files.filter(
+            function(file){ return file.match(new RegExp(playlistName + "-\\d+\\.ts")) }
+          ).map(
+            function(file){ return path.join(playlistDir, file) }
+          ));
+
+          job.playlist = playlistPath;
+
+          callbacks.success(job);
+        });
+      });
+    },
+
+    processThumbnails: function(callbacks) {
+      if (!this.parsedOpts()['thumbnail_options']) {
+        callbacks.success(this);
+        return;
+      }
+
       logger.log("Processing thumbnails for job " + this.internalId + ".");
       var thumbOpts = this.parsedOpts()['thumbnail_options'];
       var range = JobUtils.generateRangeFromThumbOpts(thumbOpts, this.duration);
-      
-      if (range) {
-        var job = this;
-        async.parallel(
-          range.map(job.execThumbJob.bind(job)),
-          function(err, results) {
-            if (err) {
-              job.lastMessage = err.message;
-              job.finalize(1);
-            } else {
-              job.finalize(0, results);
-            }
-          }
-        );
-      } else {
+
+      if (!range) {
         // no valid range
         logger.log("No valid thumbnails to process for job " + this.internalId + ". Skipping...");
-        this.finalize(0);
+        callbacks.success(this);
+        return;
       }
+
+      var job = this;
+      async.parallel(
+        range.map(job.execThumbJob.bind(job)),
+        function(err, results) {
+          if (err) {
+            job.lastMessage = err.message;
+            callbacks.error(job);
+          } else {
+            job.thumbnails = JSON.stringify(results);
+            callbacks.success(job);
+          }
+        }
+      );
     },
     execThumbJob: function(offset) {
       var job = this;
@@ -331,43 +401,44 @@ var Job = JobUtils.getDatabase().define('Job', {
         });
       }
     },
-    finalize: function(code, thumbnails) {
+    finalize: function(code) {
       var job = this;
-      if (thumbnails) job.thumbnails = JSON.stringify(thumbnails);
-      
-      if (code == 0) {
-        if (job.tmpFile) {
-          fs.rename(job.tmpFile, job.parsedOpts()['destination_file'], function (err) {
-            if (err) {
-              if ( (err.message).match(/EXDEV/) ) {
-                /*
-                  EXDEV fix, since util.pump is deprecated, using stream.pipe
-                  example from http://stackoverflow.com/questions/11293857/fastest-way-to-copy-file-in-node-js
-                */
-                try {
-                  logger.log('ffmpeg finished successfully, trying to copy across partitions');
-                  fs.createReadStream(job.tmpFile).pipe(fs.createWriteStream(job.parsedOpts()['destination_file']));
-                  job.exitHandler(code, 'ffmpeg finished succesfully.');
-                } catch (err) {
-                  logger.log(err);
-                  job.exitHandler(-1, 'ffmpeg finished succesfully, but unable to move file to different partition (' + job.parsedOpts()['destination_file'] + ').');
-                }
 
-              } else {
-                logger.log(err);
-                job.exitHandler(-1, 'ffmpeg finished succesfully, but unable to move file to destination (' + job.parsedOpts()['destination_file'] + ').');
-              }
-            } else {
+      if (code != 0) {
+        job.exitHandler(code, "ffmpeg finished with an error: '" + job.lastMessage + "' (" + code + ").");
+        return;
+      }
+
+      if (!job.tmpFile) {
+        // No tmpFile, hence no transcoding, only thumbnails
+        job.exitHandler(code, 'finished thumbnail job.');
+        return;
+      }
+
+      fs.rename(job.tmpFile, job.parsedOpts()['destination_file'], function (err) {
+        if (err) {
+          if ( (err.message).match(/EXDEV/) ) {
+            /*
+              EXDEV fix, since util.pump is deprecated, using stream.pipe
+              example from http://stackoverflow.com/questions/11293857/fastest-way-to-copy-file-in-node-js
+            */
+            try {
+              logger.log('ffmpeg finished successfully, trying to copy across partitions');
+              fs.createReadStream(job.tmpFile).pipe(fs.createWriteStream(job.parsedOpts()['destination_file']));
               job.exitHandler(code, 'ffmpeg finished succesfully.');
+            } catch (err) {
+              logger.log(err);
+              job.exitHandler(-1, 'ffmpeg finished succesfully, but unable to move file to different partition (' + job.parsedOpts()['destination_file'] + ').');
             }
-          });          
+
+          } else {
+            logger.log(err);
+            job.exitHandler(-1, 'ffmpeg finished succesfully, but unable to move file to destination (' + job.parsedOpts()['destination_file'] + ').');
+          }
         } else {
-          // No tmpFile, hence no transcoding, only thumbnails
-          job.exitHandler(code, 'finished thumbnail job.');
+          job.exitHandler(code, 'ffmpeg finished succesfully.');
         }
-      } else {
-        job.exitHandler(code, "ffmpeg finished with an error: '" + job.lastMessage + "' (" + code + ").")
-      }
+      });
     },
     toJSON: function() {
       var obj = {
@@ -377,13 +448,17 @@ var Job = JobUtils.getDatabase().define('Job', {
         'duration': this.duration,
         'filesize': this.filesize,
         'message': this.message,
-        
       };
 
       if (this.thumbnails) {
         obj['thumbnails'] = JSON.parse(this.thumbnails);
       }
-      
+
+      if (this.playlist) { obj['playlist'] = this.playlist; }
+      if (this.segments) {
+        obj['segments'] = JSON.parse(this.segments);
+      }
+
       return obj;
     },
     progressHandler: function(data) {
@@ -448,4 +523,4 @@ var Job = JobUtils.getDatabase().define('Job', {
   }
 });
 
-module.exports = Job;
+module.exports = Job;

migrations/20140822085031-add-playlist-and-segments-to-jobs.js

@@ -0,0 +1,12 @@
+module.exports = {
+  up: function(migration, DataTypes, done) {
+    // add altering commands here
+    migration.addColumn('Jobs', 'playlist', { type: DataTypes.STRING, defaultValue: null }).complete(done);
+    migration.addColumn('Jobs', 'segments', { type: DataTypes.TEXT, defaultValue: null }).complete(done);
+  },
+  down: function(migration, DataTypes, done) {
+    // add reverting commands here
+    migration.removeColumn('Jobs', 'playlist').complete(done);
+    migration.removeColumn('Jobs', 'segments').complete(done);
+  }
+}