Added further options for the test page.

Debugged subscribe/unsubscribe
prgsmall · Mar 29, 2012 · 578ca97 · 578ca97
1 parent a736ad2
commit 578ca97
Show file tree

Hide file tree

Showing 5 changed files with 65 additions and 35 deletions.
diff --git a/examples/acequiaClient.html b/examples/acequiaClient.html
@@ -18,26 +18,39 @@
                 alert('Currently connected clients: ' + message.body);
             };
 
-            var ontest = function(message, ac) {
+            var onmessage = function(message, ac) {
                 $("<div></div>")
-                    .html(message.name + " " + message.body)
+                    .html(JSON.stringify(message))
                     .appendTo("#received-messages");
             };
 
             var onconnectionChanged = function (connected) {
                 $("#connect").button("option", "disabled", connected);
-                $("#disconnect").button("option", "disabled", !connected);
                 $("#send").button("option", "disabled", !connected);
             };
 
-            var ac = new AcequiaClient("http://localhost:9091", "user" + Math.random());
-            ac.addListener(msg.MSG_CONNECT, onconnect);
-            ac.addListener(msg.MSG_GETCLIENTS, ongetClients);
+            var ac = new AcequiaClient("user" + Math.random());
+            ac.on(msg.MSG_CONNECT, onconnect);
+            ac.on(msg.MSG_GETCLIENTS, ongetClients);
+            ac.on("test", onmessage);
             ac.addConnectionChangeHandler(onconnectionChanged);
+
+            var sendMessage = function() {
+                var msgName = $("#sendMessageName").val().trim();
+                if (msgName) {
+                    ac.send(msgName);
+                }
+            };
+
+            var subscribeMessage = function() {
+                var msgName = $("#subMessageName").val().trim();
+                if (msgName) {
+                    ac.on(msgName, onmessage);
+                }
+            };
 
             $(document).ready( function(){
                 $("button").button();
-                $("#disconnect").button("option", "disabled", true);
                 $("#send").button("option", "disabled", true);
             });
         </script>
@@ -49,10 +62,10 @@
             <td><button id="connect" onclick="ac.connect();">Connect</button></td>
         </tr>
         <tr>
-            <td><button id="disconnect" onclick="ac.disconnect();">Disconnect</button></td>
+            <td><button id="send" onclick="sendMessage();">Send Message</button>&nbsp;<input id="sendMessageName" value="test"/></td>
         </tr>
         <tr>
-            <td><button id="send" onclick="ac.send('test');">Send Test Message</button></td>
+            <td><button id="subscribe" onclick="subscribeMessage();">Subscribe Message</button>&nbsp;<input id="subMessageName" value="test"/></td>
         </tr>
         <tr>
             <td><div id="received-messages" style="border:1px black solid;height:200px;width:500px;"></div></td>

diff --git a/lib/acequia.js b/lib/acequia.js
@@ -86,7 +86,7 @@ function startServers() {
             res.end(out);        
         };
 
-        console.log("HTTP server received " + req.url);
+        logger.trace("HTTP server received " + req.url);
 
         if (pathName === "/acequia/acequia.js") {
             serveClientCode(clientCode.full);
@@ -104,14 +104,14 @@ function startServers() {
     // Create the socket io server and attach it to the httpServer
     wsServer = require('socket.io').listen(httpServer);
 
-    wsServer.configure("", function () {
+    wsServer.configure(function () {
         wsServer.enable('browser client minification');  // send minified client
         wsServer.enable('browser client etag');          // apply etag caching logic based on version number
         wsServer.enable('browser client gzip');          // gzip the file
         wsServer.set('log level', 1);                    // reduce logging
         wsServer.set('transports', [                     // enable all transports (optional if you want flashsocket)
             'websocket',
-            //'flashsocket',
+            'flashsocket',
             'htmlfile',
             'xhr-polling',
             'jsonp-polling'

diff --git a/lib/acequiaClient.js b/lib/acequiaClient.js
@@ -1,4 +1,4 @@
-/*global document console io msg*/
+/*global document window console io msg*/
 
 /**
  * Creates an event callback by wrapping the object with a closure.
@@ -74,12 +74,23 @@ AcequiaClient.prototype.getSocketIOJS = function (onload) {
  * message arrives.
  */
 AcequiaClient.prototype.addListener = function (msgName, callback) {
+    var i, callbackFound = false;
+
     if (!(msgName in this.listeners)) {
         this.listeners[msgName] = [];
+        this.listeners[msgName].push(callback);
+        this.subscribes.push(msgName);
+        if (this.isConnected()) {
+            this.send(msg.MSG_SUBSCRIBE, msgName);
+        }
+    } else {
+        for (i = 0; i < this.listeners[msgName].length && !callbackFound; i += 1) {
+            callbackFound = (callback === this.listeners[msgName][i]);
+        }
+        if (!callbackFound) {
+            this.listeners[msgName].push(callback);
+        }
     }
-
-    this.listeners[msgName].push(callback);
-    this.subscribes.push(msgName);
 };
 
 /**
@@ -101,20 +112,26 @@ AcequiaClient.prototype.on = function (msgName, callback) {
 AcequiaClient.prototype.removeListener = function (msgName, callback) {
     var idx = this.listeners[msgName].indexOf(callback);
     this.listeners[msgName].splice(idx, 1);
+
+    if (this.listeners[msgName].length === 0 && this.isConnected()) {
+        this.send(msg.MSG_UNSUBSCRIBE, msgName);
+    }
 };
 
 /**
  * Connects to the Acequia server by creating a new web socket connection;
+ * @param {String} url The URL of the acequia server to connect to 
  */
-AcequiaClient.prototype.connect = function () {
-    var onload = function (that) {
+AcequiaClient.prototype.connect = function (url) {
+    var onload = function (that, uri) {
         return function (evt) {
-            that.socket = io.connect(that.uri, {reconnect: false});
+            uri = (typeof(uri) === "undefined") ? that.uri : uri;
+            that.socket = io.connect(uri, {reconnect: false});
             that.socket.on("connect", objCallback(that, "onConnect"));
         };
     };
 
-    this.getSocketIOJS(onload(this));    
+    this.getSocketIOJS(onload(this, url));    
 };
 
 /**
@@ -162,8 +179,9 @@ AcequiaClient.prototype.onConnect = function (evt) {
  * @param {Event} evt  The event object.
  */
 AcequiaClient.prototype.onDisconnect = function (evt) {
-    this.setConnected(false);
-    this.socket = null;
+//    alert("onDisconnect");
+//    this.setConnected(false);
+//    this.socket = null;
 };
 
 /**

diff --git a/lib/client.js b/lib/client.js
@@ -104,8 +104,6 @@ AcequiaClient.prototype.unsubscribeAll = function () {
 
 AcequiaClient.prototype.onMessage = function (message) {
 
-    logger.trace(message.toString());
-
     if (message.from === this.name) {
         this.update();
     } else {
@@ -259,6 +257,7 @@ AcequiaClients.prototype.createClient = function (type, message, socket, rinfo)
  */
 AcequiaClients.prototype.onMessage = function (type, message, socket, rinfo) {
     var client, i, clientList;
+    logger.trace("Received: " + message.toString());
     switch (message.name) {
     case msg.MSG_CONNECT:
         client = this.createClient(type, message, socket, rinfo);
@@ -292,14 +291,14 @@ AcequiaClients.prototype.onMessage = function (type, message, socket, rinfo) {
         this.get(message.from).send(new msg.AcequiaMessage("SYS", msg.MSG_GETCLIENTS, clientList));
         break;
 
-    case msg.MSG_SUBCRIBE:
+    case msg.MSG_SUBSCRIBE:
         client = this.get(message.from);
         for (i = 0;  i < message.body.length; i += 1) {
             client.subscribe(message.body[i]);
         }
         break;
 
-    case msg.MSG_UNSUBCRIBE:
+    case msg.MSG_UNSUBSCRIBE:
         client = this.get(message.from);
         for (i = 0;  i < message.body.length; i += 1) {
             client.unsubscribe(message.body[i]);

diff --git a/lib/msg.js b/lib/msg.js
@@ -13,15 +13,15 @@
         var message;
         if (typeof(from) === "object") {
             message = from;
-            this.from = message.from;
-            this.to   = message.to;
             this.name = message.name;
             this.body = message.body;
+            this.from = message.from;
+            this.to   = message.to;
         } else {
-            this.from = from;
-            this.to   = (typeof(to) === "undefined") ? "" : to;
             this.name = name;
             this.body = (typeof(body) === "undefined") ? [] : ((body instanceof Array) ? body : [body]);
+            this.from = from;
+            this.to   = (typeof(to) === "undefined") ? "" : to;
         }
         this.timestamp = "";
     }
@@ -60,10 +60,10 @@
     exports.AcequiaMessage = AcequiaMessage;
 
     // The standard messages that the acequia system handles.
-    exports.MSG_CONNECT      = "/connect";
-    exports.MSG_DISCONNECT   = "/disconnect";
-    exports.MSG_GETCLIENTS   = "/getClients";
-    exports.MSG_SUBSCRIBE    = "/subscribe";
-    exports.MSG_UNSUBSCRIBE  = "/unsubscribe";
+    exports.MSG_CONNECT      = "ACEQUIA_CONNECT";
+    exports.MSG_DISCONNECT   = "ACEQUIA_DISCONNECT";
+    exports.MSG_GETCLIENTS   = "ACEQUIA_GETCLIENTS";
+    exports.MSG_SUBSCRIBE    = "ACEQUIA_SUBSCRIBE";
+    exports.MSG_UNSUBSCRIBE  = "ACEQUIA_UNSUBSCRIBE";
 
 })(typeof(exports) === "undefined" ? this["msg"] = {} : exports);