message bus work in progress

config/initializers/message_bus.rb

@@ -31,3 +31,4 @@ MessageBus.is_admin_lookup do |env|
 end
 
 MessageBus.cache_assets = !Rails.env.development?
+MessageBus.enable_diagnostics

vendor/gems/message_bus/assets/application.js

@@ -1,5 +1,4 @@
 
-
 window.App = Ember.Application.createWithMixins({
   start: function(){
     MessageBus.start();
@@ -29,7 +28,7 @@ App.IndexModel = Ember.Object.extend({
     this.set("discovering", true);
     Ember.run.later(function(){
       _this.set("discovering", false);
-    }, 20 * 1000);
+    }, 1 * 1000);
 
     $.post("/message-bus/_diagnostics/discover");
 

vendor/gems/message_bus/assets/message-bus.js

@@ -30,6 +30,24 @@ window.MessageBus = (function() {
     }
   };
 
+  var processMessages = function(messages) {
+    failCount = 0;
+    $.each(messages,function(idx,message) {
+      gotData = true;
+      $.each(callbacks,function(idx,callback) {
+        if (callback.channel === message.channel) {
+          callback.last_id = message.message_id;
+          callback.func(message.data);
+        }
+        if (message["channel"] === "/__status") {
+          if (message.data[callback.channel] !== void 0) {
+            callback.last_id = message.data[callback.channel];
+          }
+        }
+      });
+    });
+  };
+
   return {
 
     enableLongPolling: true,
@@ -52,7 +70,7 @@ window.MessageBus = (function() {
           return;
         }
         data = {};
-        callbacks.each(function(c) {
+        $.each(callbacks, function(idx,c) {
           return data[c.channel] = c.last_id === void 0 ? -1 : c.last_id;
         });
         gotData = false;
@@ -65,21 +83,7 @@ window.MessageBus = (function() {
             'X-SILENCE-LOGGER': 'true'
           },
           success: function(messages) {
-            failCount = 0;
-            return messages.each(function(message) {
-              gotData = true;
-              return callbacks.each(function(callback) {
-                if (callback.channel === message.channel) {
-                  callback.last_id = message.message_id;
-                  callback.func(message.data);
-                }
-                if (message["channel"] === "/__status") {
-                  if (message.data[callback.channel] !== void 0) {
-                    callback.last_id = message.data[callback.channel];
-                  }
-                }
-              });
-            });
+            processMessages(messages);
           },
           error: failCount += 1,
           complete: function() {

vendor/gems/message_bus/lib/message_bus.rb

@@ -125,6 +125,12 @@ module MessageBus::Implementation
     @reliable_pub_sub ||= MessageBus::ReliablePubSub.new redis_config
   end
 
+  def enable_diagnostics
+    subscribe('/discover') do |msg|
+      MessageBus.publish '/process-discovery', Process.pid, user_id: msg.data[:user_id]
+    end
+  end
+
   def publish(channel, data, opts = nil) 
     return if @off 
     

vendor/gems/message_bus/lib/message_bus/rack/diagnostics.rb

@@ -71,8 +71,9 @@ HTML
 
     return index unless route
 
-    if route == 'discover'
-      MessageBus.publish('/discover', {user_id: MessageBus() })
+    if route == '/discover'
+      MessageBus.publish('/discover', {user_id: MessageBus.user_id_lookup.call(env)})
+      return [200, {}, ['ok']]
     end
 
     asset = route.split('/assets/')[1]