download file in child process

Author: Amoki

lib/helpers/child-process.js

@@ -1,13 +1,57 @@
 "use strict";
+
+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,39 +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) {
-          // 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 = new restify.BadGatewayError('Error when downloading file ' + task.file_path + ': ' + res.statusCode);
-              stream.end();
-              this.abort();
-            }
-          });
-          req.pipe(stream);
-        }
-        else {
-          cb(null);
-        }
-      },
       function performHydration(cb) {
         var child = shellFork(__dirname + '/child-process.js', {silent: true});
         var stderr = "";
@@ -86,6 +52,12 @@ module.exports = function(hydraterFunction, logger, errLogger) {
             cleaner.called = true;
             cb(err, changes);
           }
+          if(stdout !== "") {
+            logger('stdout', stdout);
+          }
+          if(stderr !== "") {
+            errLogger('stderr', stderr);
+          }
           clearTimeout(timeout);
         };
         cleaner.called = false;
@@ -125,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,
@@ -142,7 +114,6 @@ module.exports = function(hydraterFunction, logger, errLogger) {
             err = null;
           }
           cleaner(err, res.changes);
-
         });
 
         timeout = setTimeout(function() {
@@ -215,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);
     });
   };
 };

test/helpers/hydrater.js

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