Bug 1025799 - Progress events for app install. r=ochameau

toolkit/devtools/apps/app-actor-front.js

@@ -5,6 +5,7 @@ const {FileUtils} = Cu.import("resource://gre/modules/FileUtils.jsm");
 const {NetUtil} = Cu.import("resource://gre/modules/NetUtil.jsm");
 const {devtools} = Cu.import("resource://gre/modules/devtools/Loader.jsm", {});
 const {Promise: promise} = Cu.import("resource://gre/modules/Promise.jsm", {});
+const EventEmitter = require("devtools/toolkit/event-emitter");
 
 // XXX: bug 912476 make this module a real protocol.js front
 // by converting webapps actor to protocol.js
@@ -18,6 +19,9 @@ const CHUNK_SIZE = 10000;
 
 const appTargets = new Map();
 
+const AppActorFront = exports;
+EventEmitter.decorate(AppActorFront);
+
 function addDirToZip(writer, dir, basePath) {
   let files = dir.directoryEntries;
 
@@ -102,15 +106,34 @@ function uploadPackageJSON(client, webappsActor, packageFile) {
     openFile(res.actor);
   });
 
+  let fileSize;
+  let bytesRead = 0;
+
+  function emitProgress() {
+    emitInstallProgress({
+      bytesSent: bytesRead,
+      totalBytes: fileSize
+    });
+  }
+
   function openFile(actor) {
+    let openedFile;
     OS.File.open(packageFile.path)
-      .then(function (file) {
-        uploadChunk(actor, file);
+      .then(file => {
+        openedFile = file;
+        return openedFile.stat();
+      })
+      .then(fileInfo => {
+        fileSize = fileInfo.size;
+        emitProgress();
+        uploadChunk(actor, openedFile);
       });
   }
   function uploadChunk(actor, file) {
     file.read(CHUNK_SIZE)
         .then(function (bytes) {
+          bytesRead += bytes.length;
+          emitProgress();
           // To work around the fact that JSON.stringify translates the typed
           // array to object, we are encoding the typed array here into a string
           let chunk = String.fromCharCode.apply(null, bytes);
@@ -168,7 +191,11 @@ function uploadPackageBulk(client, webappsActor, packageFile) {
 
     request.on("bulk-send-ready", ({copyFrom}) => {
       NetUtil.asyncFetch(packageFile, function(inputStream) {
-        copyFrom(inputStream).then(() => {
+        let copying = copyFrom(inputStream);
+        copying.on("progress", (e, progress) => {
+          emitInstallProgress(progress);
+        });
+        copying.then(() => {
           console.log("Bulk upload done");
           inputStream.close();
           deferred.resolve(actor);
@@ -236,6 +263,16 @@ function installPackaged(client, webappsActor, packagePath, appId) {
 }
 exports.installPackaged = installPackaged;
 
+/**
+ * Emits numerous events as packaged app installation proceeds.
+ * The progress object contains:
+ *  * bytesSent:  The number of bytes sent so far
+ *  * totalBytes: The total number of bytes to send
+ */
+function emitInstallProgress(progress) {
+  AppActorFront.emit("install-progress", progress);
+}
+
 function installHosted(client, webappsActor, appId, metadata, manifest) {
   let deferred = promise.defer();
   let request = {

toolkit/devtools/apps/tests/unit/test_webappsActor.js

@@ -3,7 +3,9 @@
 
 const {devtools} = Cu.import("resource://gre/modules/devtools/Loader.jsm", {});
 const {require} = devtools;
-const {installHosted, installPackaged} = require("devtools/app-actor-front");
+const AppActorFront = require("devtools/app-actor-front");
+const {installHosted, installPackaged} = AppActorFront;
+const {Promise: promise} = Cu.import("resource://gre/modules/Promise.jsm", {});
 
 let gAppId = "actor-test";
 const APP_ORIGIN = "app://" + gAppId;
@@ -178,29 +180,54 @@ add_test(function testFileUploadInstall() {
   // Disable the bulk trait temporarily to test the JSON upload path
   gClient.traits.bulk = false;
 
-  installPackaged(gClient, gActor, packageFile.path, gAppId)
+  let progressDeferred = promise.defer();
+  // Ensure we get at least one progress event at the end
+  AppActorFront.on("install-progress", function onProgress(e, progress) {
+    if (progress.bytesSent == progress.totalBytes) {
+      AppActorFront.off("install-progress", onProgress);
+      progressDeferred.resolve();
+    }
+  });
+
+  let installed =
+    installPackaged(gClient, gActor, packageFile.path, gAppId)
     .then(function ({ appId }) {
       do_check_eq(appId, gAppId);
-
-      // Restore default bulk trait value
-      gClient.traits.bulk = true;
-
-      run_next_test();
     }, function (e) {
       do_throw("Failed install uploaded packaged app: " + e.error + ": " + e.message);
     });
+
+  promise.all([progressDeferred.promise, installed])
+    .then(() => {
+      // Restore default bulk trait value
+      gClient.traits.bulk = true;
+      run_next_test();
+    });
 });
 
 add_test(function testBulkUploadInstall() {
   let packageFile = do_get_file("data/app.zip");
   do_check_true(gClient.traits.bulk);
-  installPackaged(gClient, gActor, packageFile.path, gAppId)
+
+  let progressDeferred = promise.defer();
+  // Ensure we get at least one progress event at the end
+  AppActorFront.on("install-progress", function onProgress(e, progress) {
+    if (progress.bytesSent == progress.totalBytes) {
+      AppActorFront.off("install-progress", onProgress);
+      progressDeferred.resolve();
+    }
+  });
+
+  let installed =
+    installPackaged(gClient, gActor, packageFile.path, gAppId)
     .then(function ({ appId }) {
       do_check_eq(appId, gAppId);
-      run_next_test();
     }, function (e) {
       do_throw("Failed bulk install uploaded packaged app: " + e.error + ": " + e.message);
     });
+
+  promise.all([progressDeferred.promise, installed])
+    .then(run_next_test);
 });
 
 add_test(function testInstallHosted() {
@@ -250,4 +277,3 @@ function run_test() {
 
   run_next_test();
 }
-

toolkit/devtools/client/dbg-client.jsm

@@ -663,6 +663,8 @@ DebuggerClient.prototype = {
    *             @return Promise
    *                     The promise is resolved when copying completes or
    *                     rejected if any (unexpected) errors occur.
+   *                     This object also emits "progress" events for each chunk
+   *                     that is copied.  See stream-utils.js.
    */
   request: function (aRequest, aOnResponse) {
     if (!this.mainRoot) {
@@ -728,6 +730,8 @@ DebuggerClient.prototype = {
    *             @return Promise
    *                     The promise is resolved when copying completes or
    *                     rejected if any (unexpected) errors occur.
+   *                     This object also emits "progress" events for each chunk
+   *                     that is copied.  See stream-utils.js.
    *         * json-reply: The server replied with a JSON packet, which is
    *           passed as event data.
    *         * bulk-reply: The server replied with bulk data, which you can read
@@ -753,6 +757,8 @@ DebuggerClient.prototype = {
    *             @return Promise
    *                     The promise is resolved when copying completes or
    *                     rejected if any (unexpected) errors occur.
+   *                     This object also emits "progress" events for each chunk
+   *                     that is copied.  See stream-utils.js.
    */
   startBulkRequest: function(request) {
     if (!this.traits.bulk) {
@@ -932,6 +938,8 @@ DebuggerClient.prototype = {
    *          @return Promise
    *                  The promise is resolved when copying completes or rejected
    *                  if any (unexpected) errors occur.
+   *                  This object also emits "progress" events for each chunk
+   *                  that is copied.  See stream-utils.js.
    */
   onBulkPacket: function(packet) {
     let { actor, type, length } = packet;

toolkit/devtools/server/main.js

@@ -1199,6 +1199,8 @@ DebuggerServerConnection.prototype = {
    *          @return Promise
    *                  The promise is resolved when copying completes or rejected
    *                  if any (unexpected) errors occur.
+   *                  This object also emits "progress" events for each chunk
+   *                  that is copied.  See stream-utils.js.
    */
   onBulkPacket: function(packet) {
     let { actor: actorKey, type, length } = packet;

toolkit/devtools/transport/packets.js

@@ -28,6 +28,7 @@ const { Cc, Ci, Cu } = require("chrome");
 const DevToolsUtils = require("devtools/toolkit/DevToolsUtils");
 const { dumpn, dumpv } = DevToolsUtils;
 const StreamUtils = require("devtools/toolkit/transport/stream-utils");
+const EventEmitter = require("devtools/toolkit/event-emitter");
 
 DevToolsUtils.defineLazyGetter(this, "unicodeConverter", () => {
   const unicodeConverter = Cc["@mozilla.org/intl/scriptableunicodeconverter"]
@@ -274,8 +275,9 @@ BulkPacket.prototype.read = function(stream) {
     length: this.length,
     copyTo: (output) => {
       dumpv("CT length: " + this.length);
-      deferred.resolve(StreamUtils.copyStream(stream, output, this.length));
-      return deferred.promise;
+      let copying = StreamUtils.copyStream(stream, output, this.length);
+      deferred.resolve(copying);
+      return copying;
     },
     stream: stream,
     done: deferred
@@ -323,8 +325,9 @@ BulkPacket.prototype.write = function(stream) {
   this._readyForWriting.resolve({
     copyFrom: (input) => {
       dumpv("CF length: " + this.length);
-      deferred.resolve(StreamUtils.copyStream(input, stream, this.length));
-      return deferred.promise;
+      let copying = StreamUtils.copyStream(input, stream, this.length);
+      deferred.resolve(copying);
+      return copying;
     },
     stream: stream,
     done: deferred

toolkit/devtools/transport/stream-utils.js

@@ -8,6 +8,7 @@ const { Ci, Cc, Cu, Cr, CC } = require("chrome");
 const Services = require("Services");
 const DevToolsUtils = require("devtools/toolkit/DevToolsUtils");
 const { dumpv } = DevToolsUtils;
+const EventEmitter = require("devtools/toolkit/event-emitter");
 
 DevToolsUtils.defineLazyGetter(this, "IOUtil", () => {
   return Cc["@mozilla.org/io-util;1"].getService(Ci.nsIIOUtil);
@@ -59,6 +60,7 @@ function copyStream(input, output, length) {
 }
 
 function StreamCopier(input, output, length) {
+  EventEmitter.decorate(this);
   this._id = StreamCopier._nextId++;
   this.input = input;
   // Save off the base output stream, since we know it's async as we've required
@@ -70,13 +72,20 @@ function StreamCopier(input, output, length) {
                   createInstance(Ci.nsIBufferedOutputStream);
     this.output.init(output, BUFFER_SIZE);
   }
+  this._length = length;
   this._amountLeft = length;
   this._deferred = promise.defer();
 
   this._copy = this._copy.bind(this);
   this._flush = this._flush.bind(this);
   this._destroy = this._destroy.bind(this);
-  this._deferred.promise.then(this._destroy, this._destroy);
+
+  // Copy promise's then method up to this object.
+  // Allows the copier to offer a promise interface for the simple succeed or
+  // fail scenarios, but also emit events (due to the EventEmitter) for other
+  // states, like progress.
+  this.then = this._deferred.promise.then.bind(this._deferred.promise);
+  this.then(this._destroy, this._destroy);
 
   // Stream ready callback starts as |_copy|, but may switch to |_flush| at end
   // if flushing would block the output stream.
@@ -86,15 +95,17 @@ StreamCopier._nextId = 0;
 
 StreamCopier.prototype = {
 
-  get copied() { return this._deferred.promise; },
-
   copy: function() {
-    try {
-      this._copy();
-    } catch(e) {
-      this._deferred.reject(e);
-    }
-    return this.copied;
+    // Dispatch to the next tick so that it's possible to attach a progress
+    // event listener, even for extremely fast copies (like when testing).
+    Services.tm.currentThread.dispatch(() => {
+      try {
+        this._copy();
+      } catch(e) {
+        this._deferred.reject(e);
+      }
+    }, 0);
+    return this;
   },
 
   _copy: function() {
@@ -115,6 +126,7 @@ StreamCopier.prototype = {
     this._amountLeft -= bytesCopied;
     this._debug("Copied: " + bytesCopied +
                 ", Left: " + this._amountLeft);
+    this._emitProgress();
 
     if (this._amountLeft === 0) {
       this._debug("Copy done!");
@@ -126,6 +138,13 @@ StreamCopier.prototype = {
     this.input.asyncWait(this, 0, 0, Services.tm.currentThread);
   },
 
+  _emitProgress: function() {
+    this.emit("progress", {
+      bytesSent: this._length - this._amountLeft,
+      totalBytes: this._length
+    });
+  },
+
   _flush: function() {
     try {
       this.output.flush();

toolkit/devtools/transport/transport.js

@@ -88,6 +88,8 @@ const PACKET_HEADER_MAX = 200;
  *     @return Promise
  *             The promise is resolved when copying completes or rejected if any
  *             (unexpected) errors occur.
+ *             This object also emits "progress" events for each chunk that is
+ *             copied.  See stream-utils.js.
  *
  * - onClosed(reason) - called when the connection is closed. |reason| is
  *   an optional nsresult or object, typically passed when the transport is
@@ -172,6 +174,8 @@ DebuggerTransport.prototype = {
    *             @return Promise
    *                     The promise is resolved when copying completes or
    *                     rejected if any (unexpected) errors occur.
+   *                     This object also emits "progress" events for each chunk
+   *                     that is copied.  See stream-utils.js.
    */
   startBulkSend: function(header) {
     let packet = new BulkPacket(this);
@@ -576,9 +580,10 @@ LocalDebuggerTransport.prototype = {
         type: type,
         length: length,
         copyTo: (output) => {
-          deferred.resolve(
-            StreamUtils.copyStream(pipe.inputStream, output, length));
-          return deferred.promise;
+          let copying =
+            StreamUtils.copyStream(pipe.inputStream, output, length);
+          deferred.resolve(copying);
+          return copying;
         },
         stream: pipe.inputStream,
         done: deferred
@@ -598,9 +603,10 @@ LocalDebuggerTransport.prototype = {
 
       sendDeferred.resolve({
         copyFrom: (input) => {
-          copyDeferred.resolve(
-            StreamUtils.copyStream(input, pipe.outputStream, length));
-          return copyDeferred.promise;
+          let copying =
+            StreamUtils.copyStream(input, pipe.outputStream, length);
+          copyDeferred.resolve(copying);
+          return copying;
         },
         stream: pipe.outputStream,
         done: copyDeferred