Show progress for URL installation. Temporarily break upload installation.

lib/roles/storage/temp.js

@@ -7,11 +7,15 @@ var formidable = require('formidable')
 var Promise = require('bluebird')
 var ApkReader = require('adbkit-apkreader')
 var request = require('request')
+var progress = require('request-progress')
 var temp = require('temp')
+var zmq = require('zmq')
 
 var logger = require('../../util/logger')
 var requtil = require('../../util/requtil')
 var Storage = require('../../util/storage')
+var wire = require('../../wire')
+var wireutil = require('../../wire/util')
 
 module.exports = function(options) {
   var log = logger.createLogger('storage-temp')
@@ -19,6 +23,13 @@ module.exports = function(options) {
     , server = http.createServer(app)
     , storage = new Storage()
 
+  // Output
+  var push = zmq.socket('push')
+  options.endpoints.push.forEach(function(endpoint) {
+    log.info('Sending output to %s', endpoint)
+    push.connect(endpoint)
+  })
+
   app.set('strict routing', true)
   app.set('case sensitive routing', true)
   app.set('trust proxy', true)
@@ -27,22 +38,40 @@ module.exports = function(options) {
     log.info('Cleaning up inactive resource "%s"', id)
   })
 
-  function process(file) {
-    log.info('Processing "%s"', file.path)
+  function processFile(file) {
+    var resolver = Promise.defer()
 
-    var reader = ApkReader.readFile(file.path)
-    var manifest = reader.readManifestSync()
+    log.info('Processing file "%s"', file.path)
+
+    resolver.progress({
+      percent: 0
+    })
+
+    process.nextTick(function() {
+      try {
+        var reader = ApkReader.readFile(file.path)
+        var manifest = reader.readManifestSync()
+        resolver.resolve(manifest)
+      }
+      catch (err) {
+        resolver.reject(err)
+      }
+    })
+
+    return resolver.promise
+  }
+
+  function storeFile(file) {
     var id = storage.store(file)
-
-    return {
-      url: util.format(
+    return Promise.resolve({
+      id: id
+    , url: util.format(
         'http://%s:%s/api/v1/resources/%s'
       , options.publicIp
       , options.port
       , id
       )
-    , manifest: manifest
-    }
+    })
   }
 
   function download(url) {
@@ -57,15 +86,25 @@ module.exports = function(options) {
       resolver.reject(err)
     }
 
+    function progressListener(state) {
+      resolver.progress(state)
+    }
+
     function closeListener() {
       resolver.resolve({
         path: path
       })
     }
 
+    resolver.progress({
+      percent: 0
+    })
+
     try {
-      var dl = request(url)
-        .pipe(fs.createWriteStream(path))
+      var req = progress(request(url))
+        .on('progress', progressListener)
+
+      var save = req.pipe(fs.createWriteStream(path))
         .on('error', errorListener)
         .on('close', closeListener)
     }
@@ -74,25 +113,97 @@ module.exports = function(options) {
     }
 
     return resolver.promise.finally(function() {
-      dl.removeListener('error', errorListener)
-      dl.removeListener('close', closeListener)
+      req.removeListener('progress', progressListener)
+      save.removeListener('error', errorListener)
+      save.removeListener('close', closeListener)
     })
   }
 
   app.post('/api/v1/resources', function(req, res) {
+    function handle(fields, files) {
+      var seq = 0
+
+      function sendProgress(data, progress) {
+        if (fields.channel) {
+          push.send([
+            fields.channel
+          , wireutil.envelope(new wire.TransactionProgressMessage(
+              options.id
+            , seq++
+            , data
+            , progress
+            ))
+          ])
+        }
+      }
+
+      function sendDone(success, data, body) {
+        if (fields.channel) {
+          push.send([
+            fields.channel
+          , wireutil.envelope(new wire.TransactionDoneMessage(
+              options.id
+            , seq++
+            , success
+            , data
+            , body ? JSON.stringify(body) : null
+            ))
+          ])
+        }
+      }
+
+      if (files.file) {
+        return processFile(files.file)
+          .progressed(function(progress) {
+            sendProgress('processing', 0.9 * progress.percent)
+          })
+          .then(function(manifest) {
+            sendProgress('storing', 90)
+            return storeFile(files.file)
+              .then(function(data) {
+                data.manifest = manifest
+                sendDone(true, 'success', data)
+                return data
+              })
+          })
+          .catch(function(err) {
+            sendDone(false, 'fail')
+            return Promise.reject(err)
+          })
+      }
+      else if (fields.url) {
+        return download(fields.url)
+          .progressed(function(progress) {
+            sendProgress('uploading', 0.7 * progress.percent)
+          })
+          .then(function(file) {
+            return processFile(file)
+              .progressed(function(progress) {
+                sendProgress('processing', 70 + 0.2 * progress.percent)
+              })
+              .then(function(manifest) {
+                sendProgress('storing', 90)
+                return storeFile(file)
+                  .then(function(data) {
+                    data.manifest = manifest
+                    sendDone(true, 'success', data)
+                    return data
+                  })
+              })
+          })
+          .catch(function(err) {
+            sendDone(false, 'fail')
+            return Promise.reject(err)
+          })
+      }
+      else {
+        throw new requtil.ValidationError('"file" or "url" is required')
+      }
+    }
+
     var form = Promise.promisifyAll(new formidable.IncomingForm())
     form.parseAsync(req)
-      .spread(function(fields, files) {
-        if (files.file) {
-          return process(files.file)
-        }
-        else if (fields.url) {
-          return download(fields.url).then(process)
-        }
-        else {
-          throw new requtil.ValidationError('"file" or "url" is required')
-        }
-      })
+      .spread(handle)
       .then(function(data) {
         data.success = true
         res.json(201, data)