Merge pull request #25 from AnyFetch/restore-stream

Restore streams and download in child process

Author: Neamar

lib/helpers/child-process.js

@@ -1,13 +1,65 @@
 "use strict";
+/*
+ * A new process is spawned with this file for each task.
+ * Inter process comunication is used to send and receive message from the parent process regarding the task.
+ * This let us finely control the behavior of the hydration function.
+ * If it happens to get stuck in a subprocess or in an HTTP blackhole,
+ * the master process can simply kill the child and resume its normal operation.
+ * Without the sub-process isolation, we get stuck quite fast and get leaks everywhere.
+ */
+ 
+var crypto = require('crypto');
 var async = require("async");
+var fs = require('fs');
+var request = require('supertest');
+var url = require('url');
+var rarity = require('rarity');
+
 
 process.on('message', function(task) {
   var hydrate = require(task.functionPath);
+  var path = '/tmp/AFH-' + crypto.randomBytes(20).toString('hex');
   async.waterfall([
+    /**
+     * Download the file from task.file_path, store it in a temporary file if there is file_path
+     */
+    function downloadFile(cb) {
+      if(task.file_path) {
+        // Download the file
+        var stream = fs.createWriteStream(path);
+
+        // Store error if statusCode !== 200
+        var err;
+        stream.on("finish", function() {
+          cb(err);
+        });
+
+        var apiUrl = url.parse(task.file_path, false, true);
+        var req = request(apiUrl.protocol + "//" + apiUrl.host)
+          .get(apiUrl.path);
+
+        req.end().req.once('response', function(res) {
+          if(res.statusCode !== 200) {
+            err = 'Error when downloading file ' + task.file_path + ': ' + res.statusCode;
+            stream.end();
+            this.abort();
+          }
+        });
+        req.pipe(stream);
+      }
+      else {
+        cb(null);
+      }
+    },
     function startHydration(cb) {
       cb.urlCallback = task.options.urlCallback;
       cb.apiUrl = task.options.apiUrl;
-      hydrate(task.path, task.document, task.changes, cb);
+      hydrate(path, task.document, task.changes, cb);
+    },
+    function cleanFile(changes, cb) {
+      if(task.file_path) {
+        fs.unlink(path, rarity.carry([changes], cb));
+      }
     }
   ],
   function(err, changes) {

lib/helpers/hydrater.js

@@ -10,9 +10,7 @@
 var async = require('async');
 var shellFork = require('child_process').fork;
 var request = require('supertest');
-var crypto = require('crypto');
 var restify = require('restify');
-var fs = require('fs');
 var url = require('url');
 var rarity = require('rarity');
 var util = require('util');
@@ -39,36 +37,7 @@ module.exports = function(hydraterFunction, logger, errLogger) {
   return function(task, done) {
     logger("Starting task: " + ((task.file_path) ? task.file_path : task.document.id));
 
-    var path = '/tmp/AFH-' + crypto.randomBytes(20).toString('hex');
     async.waterfall([
-      /**
-       * Download the file from task.file_path, store it in a temporary file if there is file_path
-       */
-      function downloadFile(cb) {
-        if(task.file_path) {
-          var apiUrl = url.parse(task.file_path, false, true);
-          request(apiUrl.protocol + "//" + apiUrl.host)
-            .get(apiUrl.path)
-            .expect(200)
-            .end(function(err, res) {
-              if(err) {
-                err = new restify.BadGatewayError('Error when downloading file ' + task.file_path + ': ' + err);
-              }
-              cb(err, res && res.text);
-            });
-        }
-        else {
-          cb(null);
-        }
-      },
-      function saveFile(res, cb) {
-        if(res) {
-          fs.writeFile(path, res, cb);
-        }
-        else {
-          cb();
-        }
-      },
       function performHydration(cb) {
         var child = shellFork(__dirname + '/child-process.js', {silent: true});
         var stderr = "";
@@ -83,6 +52,12 @@ module.exports = function(hydraterFunction, logger, errLogger) {
             cleaner.called = true;
             cb(err, changes);
           }
+          if(stdout !== "") {
+            logger(stdout);
+          }
+          if(stderr !== "") {
+            errLogger(stderr);
+          }
           clearTimeout(timeout);
         };
         cleaner.called = false;
@@ -122,7 +97,7 @@ module.exports = function(hydraterFunction, logger, errLogger) {
 
         child.send({
           functionPath: hydraterFunction,
-          path: (task.file_path) ? path : null,
+          file_path: task.file_path,
           document: task.document,
           changes: lib.defaultChanges(),
           options: options,
@@ -139,7 +114,6 @@ module.exports = function(hydraterFunction, logger, errLogger) {
             err = null;
           }
           cleaner(err, res.changes);
-
         });
 
         timeout = setTimeout(function() {
@@ -212,21 +186,13 @@ module.exports = function(hydraterFunction, logger, errLogger) {
         if(task.next) {
           task.next(new restify.InvalidContentError("ERR hydrating " + ((task.file_path) ? task.file_path : task.document.id) + err.toString()));
         }
-        errLogger("ERR hydrating " + ((task.file_path) ? task.file_path : task.document.id), err);
+        errLogger("ERR hydrating " + ((task.file_path) ? task.file_path : task.document.id), err.toString());
       }
 
       if(res && res.statusCode && res.statusCode !== 204) {
         errLogger("ERR hydrating: server refused data! Code:" + res.statusCode);
       }
-
-      if(task.file_path) {
-        fs.unlink(path, function(_err) {
-          done(err || _err, changes);
-        });
-      }
-      else {
-        done(changes);
-      }
+      done(err, changes);
     });
   };
 };

lib/index.js

@@ -20,7 +20,7 @@ module.exports.createServer = function(config) {
     throw new Error("Specify `hydrater_function`");
   }
   config.logger = config.logger || console.log;
-  config.errLogger = config.errLogger || console.warn;
+  config.errLogger = config.errLogger || console.error;
 
   // Load configuration and initialize server
   var hydraterEndpoint = require('./handlers/hydrater.js');

test/helpers/hydrater.js

@@ -36,7 +36,7 @@ describe("hydrate()", function() {
 
       hydrate(task, function(err, changes) {
         changes.should.have.keys(["metadata"]);
-        done();
+        done(err);
       });
     });