Fix race condition in processing of messages

5 files changed, 47 insertions, 16 deletions

diff --git a/src/cljs/gnunet_web/peerinfo.cljs b/src/cljs/gnunet_web/peerinfo.cljs
index 8532209..c041dc3 100644
--- a/src/cljs/gnunet_web/peerinfo.cljs
+++ b/src/cljs/gnunet_web/peerinfo.cljs
@@ -69,7 +69,7 @@
   (encode-message {:message-type message-type-peerinfo-info-end}))
 
 (def peerinfo-message-channel (js/MessageChannel.))
-(client-connect "peerinfo" (.-port2 peerinfo-message-channel))
+(client-connect "peerinfo" "web page" (.-port2 peerinfo-message-channel))
 
 (defn process-hello
   [hello]
diff --git a/src/cljs/gnunet_web/service.cljs b/src/cljs/gnunet_web/service.cljs
index 896079f..9acf947 100644
--- a/src/cljs/gnunet_web/service.cljs
+++ b/src/cljs/gnunet_web/service.cljs
@@ -58,6 +58,7 @@
                                (fn [event]
                                  (println worker-name " err:" (.-data event)))))
                 "client_connect" (client-connect (.-service_name data)
+                                                 (.-client_name data)
                                                  (.-message_port data))
                 (println worker-name ":" (js/JSON.stringify data))))))
     (.start port)
@@ -67,14 +68,15 @@
     worker))
 
 (defn client-connect
-  [service-name message-port]
+  [service-name client-name message-port]
   (let [service (get @services service-name)]
     (if (nil? service)
       (let [worker (start-worker service-name
                                  (str "js/gnunet-service-" service-name ".js"))
             port (.-port worker)]
         (add-service service-name port)
-        (recur service-name message-port))
+        (recur service-name client-name message-port))
       (.postMessage service (clj->js {:type "connect"
+                                      :client-name client-name
                                       :port message-port})
                     (array message-port)))))
diff --git a/src/js/client.js b/src/js/client.js
index 6f996d5..adc0113 100644
--- a/src/js/client.js
+++ b/src/js/client.js
@@ -23,13 +23,15 @@ mergeInto(LibraryManager.library, {
     var channel = new MessageChannel();
     var port = NEXT_PORT;
     NEXT_PORT = port + 1;
-    CLIENT_PORTS[port] = channel.port1;
+    CLIENT_PORTS[port] = {port: channel.port1, name: service_name};
     client_connect(service_name, channel.port2);
     return port;
   },
   GNUNET_CLIENT_receive__deps: ['$CLIENT_PORTS'],
   GNUNET_CLIENT_receive: function(client, handler, handler_cls, timeout) {
-    CLIENT_PORTS[client].onmessage = function(ev) {
+    CLIENT_PORTS[client].port.onmessage = function(ev) {
+      Module.print('Received ' + ev.data.length + ' bytes from service '
+        + CLIENT_PORTS[client].name);
       ccallFunc(Runtime.getFuncWrapper(handler, 'vii'), 'void',
         ['number', 'array'],
         [handler_cls, ev.data]);
@@ -37,7 +39,7 @@ mergeInto(LibraryManager.library, {
     var delay = getValue(timeout, 'i64');
     if (-1 != delay) {
       setTimeout(function() {
-        CLIENT_PORTS[client].onmessage = null;
+        CLIENT_PORTS[client].port.onmessage = null;
         Runtime.dynCall('vii', handler, [handler_cls, 0]);
       }, delay / 1000);
     }
@@ -48,13 +50,17 @@ mergeInto(LibraryManager.library, {
     // Supposedly we can call notify right now, but the current code never
     // does so let's emulate that.
     setTimeout(function() {
-      //Module.print('I want to send ' + size + ' bytes to service ' + client);
+      //Module.print('I want to send ' + size + ' bytes to service '
+      //  + CLIENT_PORTS[client].name);
       var stack = Runtime.stackSave();
       var buffer = Runtime.stackAlloc(size);
       var ret = Runtime.dynCall('iiii', notify, [notify_cls, size, buffer]);
+      //Module.print('I\'m sending ' + size + ' bytes to service '
+      //          + CLIENT_PORTS[client].name);
       var view = {{{ makeHEAPView('U8', 'buffer', 'buffer+ret') }}};
       // See http://code.google.com/p/chromium/issues/detail?id=169705
-      CLIENT_PORTS[client].postMessage(new Uint8Array(view));
+      if (ret > 0)
+        CLIENT_PORTS[client].port.postMessage(new Uint8Array(view));
       Runtime.stackRestore(stack);
     }, 0);
     return 1; // opaque GNUNET_CLIENT_TransmitHandle*
diff --git a/src/js/pre.js b/src/js/pre.js
index fc7e032..dd2c4a5 100644
--- a/src/js/pre.js
+++ b/src/js/pre.js
@@ -121,16 +121,19 @@ function get_message(ev) {
     random_offset = 0;
     removeRunDependency('window-init');
   } else if ('connect' == ev.data.type) {
-    SERVER.connect(ev.data.port);
+    SERVER.connect(ev.data.port, ev.data['client-name']);
   }
 }
 
 // Ask a window to connect us to a service
 function client_connect(service_name, message_port) {
+  //Module.print('I want to connect to ' + service_name);
   do_to_window(function(w) {
+    //Module.print('I am now connecting to ' + service_name);
     w.postMessage({
       type: 'client_connect',
       service_name: service_name,
+      client_name: location.pathname,
       message_port: message_port}, [message_port]);
   });
 }
diff --git a/src/js/server.js b/src/js/server.js
index a7bcb8f..9863813 100644
--- a/src/js/server.js
+++ b/src/js/server.js
@@ -17,15 +17,18 @@
 mergeInto(LibraryManager.library, {
   $SERVER: {
     handlers: {},
+    handlers_initialized: false,
     connect_notify_list: [],
     disconnect_notify_list: [],
     clients: {},
     next_client: 1,
-    connect: function(port) {
+    connect: function(port, client_name) {
+      Module.print('Got a connection from ' + client_name);
       port.onmessage = SERVER.client_get_message;
       port._name = SERVER.next_client++;
       SERVER.clients[port._name] = {
         port: port,
+        name: client_name,
         ref_count: 0,
         shtudown_now: false,
       };
@@ -35,21 +38,27 @@ mergeInto(LibraryManager.library, {
           [notify.callback_cls, port._name]);
       });
     },
+    message_queue: [],
     client_get_message: function(ev) {
       var size = ev.data[0] << 8 | ev.data[1];
       var type = ev.data[2] << 8 | ev.data[3];
+      Module.print('Got message of type ' + type + ' size ' + size + ' from '
+          + SERVER.clients[ev.target._name].name);
+      if (!SERVER.handlers_initialized) {
+        Module.print('And I\'m queueing it for later');
+        SERVER.message_queue.push(ev);
+        return;
+      }
       var handler = SERVER.handlers[type];
-      //Module.print("Got message of type " + type + " size " + size + " from "
-      //    + ev.target._name);
       if (typeof handler === 'undefined') {
-        //Module.print("But I don't know what to do with it");
+        Module.print("But I don't know what to do with it");
       } else {
         if (handler.expected_size == 0 || handler.expected_size == size) {
           ccallFunc(Runtime.getFuncWrapper(handler.callback, 'viii'), 'void',
               ['number', 'number', 'array'],
               [handler.callback_cls, ev.target._name, ev.data]);
         } else {
-          //Module.print("But I was expecting size " + handler.expected_size);
+          Module.print("But I was expecting size " + handler.expected_size);
         }
       }
     }
@@ -70,6 +79,13 @@ mergeInto(LibraryManager.library, {
         'expected_size': expected_size
       };
     }
+    setTimeout(function() {
+      var queue = SERVER.message_queue;
+      SERVER.message_queue = [];
+      SERVER.handlers_initialized = true;
+      Module.print('Processing ' + queue.length + ' queued messages');
+      queue.forEach(SERVER.client_get_message);
+    }, 0);
   },
   GNUNET_SERVER_receive_done__deps: [
     '$SERVER',
@@ -107,13 +123,17 @@ mergeInto(LibraryManager.library, {
   GNUNET_SERVER_notify_transmit_ready: function(client, size, timeout, callback,
                                            callback_cls) {
     setTimeout(function() {
-      //Module.print('I want to send ' + size + ' bytes to client ' + client);
+      //Module.print('I want to send ' + size + ' bytes to client '
+      //  + SERVER.clients[client].name);
       var stack = Runtime.stackSave();
       var buffer = Runtime.stackAlloc(size);
       var ret = Runtime.dynCall('iiii', callback, [callback_cls, size, buffer]);
       var view = {{{ makeHEAPView('U8', 'buffer', 'buffer+ret') }}};
+      //Module.print('I\'m sending ' + size + ' bytes to client '
+      //  + SERVER.clients[client].name);
       // See http://code.google.com/p/chromium/issues/detail?id=169705
-      SERVER.clients[client].port.postMessage(new Uint8Array(view));
+      if (ret > 0)
+        SERVER.clients[client].port.postMessage(new Uint8Array(view));
       Runtime.stackRestore(stack);
     }, 0);
     return 1; // opaque GNUNET_SERVER_TransmitHandle*