add behavior tests to javascript client

misc/Rakefile

@@ -25,12 +25,12 @@ end
 begin
   load 'jasmine/tasks/jasmine.rake'
 
-  task "jasmine:require" => ["jshint", "configure_jasmine", "monitor_js"]
+  task "jasmine:require" => ["jshint", "configure_jasmine", "monitor_js", "test_server"]
 
   task :configure_jasmine do
     Jasmine.configure do |config|
       config.spec_dir = project_dir
-      config.spec_files = lambda { Dir["#{project_dir}/misc/spec/javascripts/helpers/**/*.js"] + Dir["#{project_dir}/misc/**/*[sS]pec.js"] }
+      config.spec_files = lambda { Dir["#{project_dir}/misc/spec/javascripts/helpers/**/*.js"] + Dir["#{project_dir}/misc/js/jquery.min.js"] + Dir["#{project_dir}/misc/**/*[sS]pec.js"] }
       js_tmp_dir = File.expand_path('pushstream/js', Dir.tmpdir)
       config.src_dir = js_tmp_dir
       config.src_files = lambda { Dir["#{js_tmp_dir}/**/*.js"] }
@@ -54,6 +54,13 @@ begin
     listener.start(false)
   end
 
+  task :test_server do
+    require File.expand_path('misc/spec/spec_helper', project_dir)
+    include NginxTestHelper
+    config = NginxTestHelper::Config.new "jasmine", {:configuration_template => File.read(File.expand_path('misc/nginx.conf', project_dir))}
+    start_server(config)
+  end
+
 rescue LoadError
   desc "Run javascript tests"
   task :jasmine do

misc/js/pushstream.js

@@ -92,8 +92,12 @@
     throw "Invalid JSON: " + data;
   };
 
+  var getTime = function() {
+    return (new Date()).getTime();
+  };
+
   var currentTimestampParam = function() {
-    return { "_" : (new Date()).getTime() };
+    return { "_" : getTime() };
   };
 
   var objectToUrlParams = function(settings) {
@@ -245,9 +249,7 @@
     },
 
     _clear_timeout : function(settings) {
-      if (settings.timeoutId) {
-        settings.timeoutId = window.clearTimeout(settings.timeoutId);
-      }
+      settings.timeoutId = clearTimer(settings.timeoutId);
     },
 
     clear : function(settings) {
@@ -261,11 +263,13 @@
 
       var head = document.head || document.getElementsByTagName("head")[0];
       var script = document.createElement("script");
-      var startTime = new Date().getTime();
+      var startTime = getTime();
 
       var onerror = function() {
         Ajax.clear(settings);
-        var endTime = new Date().getTime();
+        var callbackFunctionName = settings.data.callback;
+        if (callbackFunctionName) { window[callbackFunctionName] = function() { window[callbackFunctionName] = null; }; }
+        var endTime = getTime();
         settings.error(((endTime - startTime) > settings.timeout/2) ? 304 : 0);
       };
 
@@ -285,7 +289,12 @@
       if (settings.beforeSend) { settings.beforeSend({}); }
 
       settings.timeoutId = window.setTimeout(onerror, settings.timeout + 2000);
-      settings.scriptId = settings.scriptId || new Date().getTime();
+      settings.scriptId = settings.scriptId || getTime();
+
+      var callbackFunctionName = settings.data.callback;
+      if (callbackFunctionName) { window[callbackFunctionName] = function() { window[callbackFunctionName] = null; }; }
+      settings.data.callback = settings.scriptId + "_onmessage_" + getTime();
+      window[settings.data.callback] = settings.success;
 
       script.setAttribute("src", addParamsToUrl(settings.url, extend({}, settings.data, currentTimestampParam())));
       script.setAttribute("async", "async");
@@ -293,6 +302,7 @@
 
       // Use insertBefore instead of appendChild to circumvent an IE6 bug.
       head.insertBefore(script, head.firstChild);
+      return settings;
     },
 
     load : function(settings) {
@@ -401,7 +411,7 @@
 
   var clearTimer = function(timer) {
     if (timer) {
-      clearTimeout(timer);
+      window.clearTimeout(timer);
     }
     return null;
   };
@@ -428,7 +438,7 @@
       return;
     }
     this._closeCurrentConnection();
-    this.pushstream._onerror({type: ((event && (event.type === "load")) || (this.pushstream.readyState === PushStream.CONNECTING)) ? "load" : "timeout"});
+    this.pushstream._onerror({type: ((event && ((event.type === "load") || (event.type === "close"))) || (this.pushstream.readyState === PushStream.CONNECTING)) ? "load" : "timeout"});
   };
 
   /* wrappers */
@@ -612,7 +622,7 @@
     process: function(id, channel, text, eventid) {
       this.pingtimer = clearTimer(this.pingtimer);
       Log4js.info("[Stream] message received", arguments);
-      this.pushstream._onmessage(unescapeText(text), id, channel, eventid, true);
+      this.pushstream._onmessage(unescapeText(text), id, channel, eventid || "", true);
       this.setPingTimer();
     },
 
@@ -662,15 +672,14 @@
       var domain = Utils.extract_xss_domain(this.pushstream.host);
       var currentDomain = Utils.extract_xss_domain(window.location.hostname);
       var port = this.pushstream.port;
-      var currentPort = window.location.port || (this.pushstream.useSSL ? 443 : 80);
+      var currentPort = window.location.port ? Number(window.location.port) : (this.pushstream.useSSL ? 443 : 80);
       this.useJSONP = (domain !== currentDomain) || (port !== currentPort) || this.pushstream.useJSONP;
       this.xhrSettings.scriptId = "PushStreamManager_" + this.pushstream.id;
       if (this.useJSONP) {
         this.pushstream.messagesControlByArgument = true;
-        this.xhrSettings.data.callback = "PushStreamManager[" + this.pushstream.id + "].wrapper.onmessage";
       }
       this._internalListen();
-      this.opentimer = window.setTimeout(linker(onopenCallback, this), 5000);
+      this.opentimer = window.setTimeout(linker(onopenCallback, this), 100);
       Log4js.info("[LongPolling] connecting to:", this.xhrSettings.url);
     },
 
@@ -683,7 +692,7 @@
       if (this.connectionEnabled) {
         this.xhrSettings.data = extend({}, this.pushstream.extraParams(), this.xhrSettings.data);
         if (this.useJSONP) {
-          Ajax.jsonp(this.xhrSettings);
+          this.connection = Ajax.jsonp(this.xhrSettings);
         } else if (!this.connection) {
           this.connection = Ajax.load(this.xhrSettings);
         }
@@ -702,7 +711,9 @@
     _closeCurrentConnection: function() {
       this.opentimer = clearTimer(this.opentimer);
       if (this.connection) {
-        try { this.connection.abort(); } catch (e) { /* ignore error on closing */ }
+        try { this.connection.abort(); } catch (e) {
+          try { Ajax.clear(this.connection); } catch (e) { /* ignore error on closing */ }
+        }
         this.connection = null;
         this.lastModified = null;
         this.xhrSettings.url = null;
@@ -744,7 +755,7 @@
         } else {
           Log4js.info("[LongPolling] error (disconnected by server):", status);
           this._closeCurrentConnection();
-          this.pushstream._onerror({type: (status === 403) ? "load" : "timeout"});
+          this.pushstream._onerror({type: ((status === 403) || (this.pushstream.readyState === PushStream.CONNECTING)) ? "load" : "timeout"});
         }
       }
     },
@@ -788,7 +799,7 @@
 
     this.useSSL = settings.useSSL || false;
     this.host = settings.host || window.location.hostname;
-    this.port = settings.port || (this.useSSL ? 443 : 80);
+    this.port = Number(settings.port || (this.useSSL ? 443 : 80));
 
     this.timeout = settings.timeout || 30000;
     this.pingtimeout = settings.pingtimeout || 30000;
@@ -824,13 +835,13 @@
     this.onmessage = settings.onmessage || null;
     this.onerror = settings.onerror || null;
     this.onstatuschange = settings.onstatuschange || null;
+    this.extraParams    = settings.extraParams    || function() { return {}; };
 
     this.channels = {};
     this.channelsCount = 0;
     this.channelsByArgument   = settings.channelsByArgument   || false;
     this.channelsArgument     = settings.channelsArgument     || 'channels';
 
-    this.extraParams          = settings.extraParams          || this.extraParams;
 
     for (var i = 0; i < this.modes.length; i++) {
       try {
@@ -859,10 +870,6 @@
 
   /* main code */
   PushStream.prototype = {
-    extraParams: function() {
-      return {};
-    },
-
     addChannel: function(channel, options) {
       if (escapeText(channel) !== channel) {
         throw "Invalid channel name! Channel has to be a set of [a-zA-Z0-9]";
@@ -964,7 +971,7 @@
       Log4js.debug("message", text, id, channel, eventid, isLastMessageFromBatch);
       if (id === -2) {
         if (this.onchanneldeleted) { this.onchanneldeleted(channel); }
-      } else if ((id > 0) && (typeof(this.channels[channel]) !== "undefined")) {
+      } else if (id > 0) {
         if (this.onmessage) { this.onmessage(text, id, channel, eventid, isLastMessageFromBatch); }
       }
     },

misc/nginx.conf

@@ -15,7 +15,6 @@ events {
 }
 
 http {
-    include         mime.types;
     default_type    application/octet-stream;
 
     access_log      logs/nginx-http_access.log;
@@ -57,6 +56,7 @@ http {
     push_stream_authorized_channels_only        off;
     push_stream_wildcard_channel_max_qtd        3;
 
+    push_stream_allowed_origins                 "*";
 
     server {
         listen           9080 default_server;
@@ -96,6 +96,9 @@ http {
 
             # positional channel path
             push_stream_channels_path                   $1;
+            if ($arg_tests = "on") {
+              push_stream_channels_path                 "test_$1";
+            }
 
             # header to be sent when receiving new subscriber connection
             push_stream_header_template                 "<html><head><meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\">\r\n<meta http-equiv=\"Cache-Control\" content=\"no-store\">\r\n<meta http-equiv=\"Cache-Control\" content=\"no-cache\">\r\n<meta http-equiv=\"Pragma\" content=\"no-cache\">\r\n<meta http-equiv=\"Expires\" content=\"Thu, 1 Jan 1970 00:00:00 GMT\">\r\n<script type=\"text/javascript\">\r\nwindow.onError = null;\r\ntry{ document.domain = (window.location.hostname.match(/^(\d{1,3}\.){3}\d{1,3}$/)) ? window.location.hostname : window.location.hostname.split('.').slice(-1 * Math.max(window.location.hostname.split('.').length - 1, (window.location.hostname.match(/(\w{4,}\.\w{2}|\.\w{3,})$/) ? 2 : 3))).join('.');}catch(e){}\r\nparent.PushStream.register(this);\r\n</script>\r\n</head>\r\n<body>";
@@ -115,6 +118,9 @@ http {
 
             # positional channel path
             push_stream_channels_path                   $1;
+            if ($arg_tests = "on") {
+              push_stream_channels_path                 "test_$1";
+            }
         }
 
         location ~ /lp/(.*) {
@@ -123,6 +129,27 @@ http {
 
             # positional channel path
             push_stream_channels_path         $1;
+            if ($arg_tests = "on") {
+              push_stream_channels_path                 "test_$1";
+            }
+        }
+
+        location ~ /jsonp/(.*) {
+            # activate long-polling mode for this location
+            push_stream_subscriber      long-polling;
+
+            push_stream_longpolling_connection_ttl        1s;
+
+            push_stream_message_template                "{\"id\":~id~,\"channel\":\"~channel~\",\"text\":\"~text~\", \"tag\":\"~tag~\", \"time\":\"~time~\"}";
+
+            push_stream_last_received_message_time "$arg_time";
+            push_stream_last_received_message_tag  "$arg_tag";
+
+            # positional channel path
+            push_stream_channels_path         $1;
+            if ($arg_tests = "on") {
+              push_stream_channels_path                 "test_$1";
+            }
         }
 
         location ~ /ws/(.*) {
@@ -131,11 +158,18 @@ http {
 
             # positional channel path
             push_stream_channels_path                   $1;
+            if ($arg_tests = "on") {
+              push_stream_channels_path                 "test_$1";
+            }
 
             # store messages in memory
             push_stream_store_messages              on;
 
             push_stream_websocket_allow_publish     on;
         }
+
+        location / {
+            proxy_pass "http://localhost:8888";
+        }
     }
 }

misc/spec/javascripts/PushStreamSpec.js

 describe("PushStream", function() {
-
   beforeEach(function() {
   });
 
@@ -258,4 +257,294 @@ describe("PushStream", function() {
       expect(pushstream.wrappers[1].type).toBe("LongPolling");
     });
   });
+
+  function itShouldHaveCommonBehavior(mode, useJSONP) {
+    var pushstream = null;
+    var channelName = null;
+    var port = 9080;
+    var nginxServer = "localhost:" + port;
+    var jsonp = useJSONP || false;
+    var urlPrefixLongpolling = useJSONP ? '/jsonp' : '/lp';
+
+    beforeEach(function() {
+      for (var i = 0; i < PushStreamManager.length; i++) {
+        PushStreamManager[i].disconnect();
+      }
+      channelName = "ch_" + new Date().getTime();
+    });
+
+    afterEach(function() {
+      if (pushstream) { pushstream.disconnect(); }
+    });
+
+    describe("when connecting", function() {
+      it("should call onstatuschange callback", function() {
+        var status = [];
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixLongpolling: urlPrefixLongpolling,
+          onstatuschange: function(st) {
+            status.push(st);
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+        });
+
+        waitsFor(function() {
+          return status.length >= 2;
+        }, "The callback was not called", 1000);
+
+        runs(function() {
+          expect(status).toEqual([PushStream.CONNECTING, PushStream.OPEN]);
+        });
+      });
+    });
+
+    describe("when receiving a message", function() {
+      it("should call onmessage callback", function() {
+        var receivedMessage = false;
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixLongpolling: urlPrefixLongpolling,
+          onmessage: function(text, id, channel, eventid, isLastMessageFromBatch) {
+            expect(text).toBe("a test message");
+            expect(id).toBe(1);
+            expect(channel).toBe(channelName);
+            expect(eventid).toBe("");
+            expect(isLastMessageFromBatch).toBeTruthy();
+            receivedMessage = true;
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+
+          setTimeout(function() {
+            $.post("http://" + nginxServer + "/pub?id=" + channelName, "a test message");
+          }, 500);
+        });
+
+        waitsFor(function() {
+          return receivedMessage;
+        }, "The callback was not called", 1000);
+      });
+    });
+
+    describe("when disconnecting", function() {
+      it("should call onstatuschange callback with CLOSED status", function() {
+        var status = null;
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixLongpolling: urlPrefixLongpolling,
+          onstatuschange: function(st) {
+            status = st;
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+
+          setTimeout(function() {
+            pushstream.disconnect();
+          }, 500);
+        });
+
+        waitsFor(function() {
+          return status == PushStream.CLOSED;
+        }, "The callback was not called", 1000);
+
+        runs(function() {
+          expect(pushstream.readyState).toBe(PushStream.CLOSED);
+        });
+      });
+    });
+
+    describe("when adding a new channel", function() {
+      it("should reconnect", function() {
+        var status = [];
+        var messages = [];
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixLongpolling: urlPrefixLongpolling,
+          onstatuschange: function(st) {
+            status.push(st);
+          },
+          onmessage: function(text, id, channel, eventid, isLastMessageFromBatch) {
+            messages.push(arguments);
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+
+          setTimeout(function() {
+            pushstream.addChannel("other_" + channelName);
+          }, 200);
+        });
+
+        waitsFor(function() { return pushstream.channelsCount >= 2; }, "Channel not added", 300);
+        runs(function() {
+          $.post("http://" + nginxServer + "/pub?id=" + channelName, "a test message", function() {
+            setTimeout(function() {
+              $.post("http://" + nginxServer + "/pub?id=" + "other_" + channelName, "message on other channel");
+            }, 700);
+          });
+        });
+
+        waitsFor(function() {
+          return messages.length >= 2;
+        }, "The callback was not called", 2000);
+
+        runs(function() {
+          expect(status).toEqual([PushStream.CONNECTING, PushStream.OPEN, PushStream.CLOSED, PushStream.CONNECTING, PushStream.OPEN]);
+          expect(messages).toEqual([
+            ["a test message", 1, channelName, "", true],
+            ["message on other channel", 1, "other_" + channelName, "", true]
+          ]);
+        });
+      });
+    });
+
+    describe("when deleting a channel", function() {
+      it("should call onchanneldeleted callback", function() {
+        var channel = null;
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixLongpolling: urlPrefixLongpolling,
+          onchanneldeleted: function(ch) {
+            channel = ch;
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+
+          setTimeout(function() {
+             $.ajax({type: "DELETE", url: "http://" + nginxServer + "/pub?id=" + channelName});
+          }, 500);
+        });
+
+        waitsFor(function() {
+          return channel !== null;
+        }, "The callback was not called", 1000);
+
+        runs(function() {
+          $.post("http://" + nginxServer + "/pub?id=" + channelName, "a test message", function() {
+            $.ajax({
+              url: "http://" + nginxServer + "/pub?id=" + channelName,
+              success: function(data) {
+                expect(data.published_messages).toBe("1");
+              }
+            });
+          });
+          expect(channel).toBe(channelName);
+        });
+      });
+    });
+
+    describe("when sending extra params", function() {
+      it("should call extraParams function", function() {
+        var receivedMessage = false;
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixLongpolling: urlPrefixLongpolling,
+          extraParams: function() {
+            return {"tests":"on"};
+          },
+          onmessage: function(text, id, channel, eventid, isLastMessageFromBatch) {
+            expect(text).toBe("a test message");
+            expect(id).toBe(1);
+            expect(channel).toBe("test_" + channelName);
+            expect(eventid).toBe("");
+            expect(isLastMessageFromBatch).toBeTruthy();
+            receivedMessage = true;
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+
+          setTimeout(function() {
+            $.post("http://" + nginxServer + "/pub?id=" + "test_" + channelName, "a test message");
+          }, 500);
+        });
+
+        waitsFor(function() {
+          return receivedMessage;
+        }, "The callback was not called", 1000);
+      });
+    });
+
+    describe("when an error on connecting happens", function() {
+      it("should call onerror callback with a load error type", function() {
+        var error = null;
+        pushstream = new PushStream({
+          modes: mode,
+          port: port,
+          useJSONP: jsonp,
+          urlPrefixStream: '/pub',
+          urlPrefixEventsource: '/pub',
+          urlPrefixLongpolling: '/pub',
+          urlPrefixWebsocket: '/pub',
+          onerror: function(err) {
+            error = err;
+          }
+        });
+        pushstream.addChannel(channelName);
+
+        runs(function() {
+          pushstream.connect();
+        });
+
+        waitsFor(function() {
+          return error !== null;
+        }, "The callback was not called", 1000);
+
+        runs(function() {
+          expect(pushstream.readyState).toBe(PushStream.CLOSED);
+          expect(error.type).toBe("load");
+        });
+      });
+    });
+  };
+
+  describe("on Stream mode", function() {
+    itShouldHaveCommonBehavior('stream');
+  });
+
+  describe("on EventSource mode", function() {
+    itShouldHaveCommonBehavior('eventsource');
+  });
+
+  describe("on WebSocket mode", function() {
+    itShouldHaveCommonBehavior('websocket');
+  });
+
+  describe("on LongPolling mode", function() {
+    itShouldHaveCommonBehavior('longpolling');
+  });
+
+  describe("on JSONP mode", function() {
+    itShouldHaveCommonBehavior('longpolling', true);
+  });
 });