activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject [2/2] git commit: Fixed bug https://issues.apache.org/jira/browse/APLO-355
Date Tue, 29 Apr 2014 14:02:52 GMT
Fixed bug https://issues.apache.org/jira/browse/APLO-355


Project: http://git-wip-us.apache.org/repos/asf/activemq-apollo/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-apollo/commit/1e3eadff
Tree: http://git-wip-us.apache.org/repos/asf/activemq-apollo/tree/1e3eadff
Diff: http://git-wip-us.apache.org/repos/asf/activemq-apollo/diff/1e3eadff

Branch: refs/heads/trunk
Commit: 1e3eadff46fa4b3bb2ca2ae7ec905e831b83818c
Parents: 35c3c79
Author: leiqin <leiqin2010@gmail.com>
Authored: Mon Apr 28 16:29:26 2014 +0800
Committer: leiqin <leiqin2010@gmail.com>
Committed: Mon Apr 28 16:29:26 2014 +0800

----------------------------------------------------------------------
 .../jetty/WebSocketTransportFactory.scala       |   8 +-
 .../examples/stomp/websocket/js/stomp.js        | 207 ++++++++++++++-----
 2 files changed, 155 insertions(+), 60 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-apollo/blob/1e3eadff/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
----------------------------------------------------------------------
diff --git a/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
b/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
index d3fb727..df921f4 100644
--- a/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
+++ b/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
@@ -36,7 +36,7 @@ import java.nio.channels._
 import scala.collection.mutable.ListBuffer
 import java.util.concurrent.{ExecutorService, Executor, ArrayBlockingQueue}
 import org.fusesource.hawtdispatch.transport.ProtocolCodec.BufferState
-import org.fusesource.hawtbuf.{AsciiBuffer, Buffer}
+import org.fusesource.hawtbuf.{UTF8Buffer, AsciiBuffer, Buffer}
 import java.io.{EOFException, IOException}
 import java.security.cert.X509Certificate
 import org.apache.activemq.apollo.broker.web.AllowAnyOriginFilter
@@ -366,7 +366,7 @@ object WebSocketTransportFactory extends TransportFactory.Provider with
Log {
         first_message = false
       }
       // Convert string messages to bytes messages..  our codecs just work with bytes..
-      var buffer = new AsciiBuffer(str)
+      var buffer = new UTF8Buffer(str)
       onMessage(buffer.data, buffer.offset, buffer.length)
     }
 
@@ -597,7 +597,7 @@ object WebSocketTransportFactory extends TransportFactory.Provider with
Log {
           if( service_state.is_starting_or_started || !write_failed) {
             try {
               if (!binary_transfers) {
-                connection.sendMessage(buffer.ascii().toString)
+                connection.sendMessage(buffer.utf8().toString)
               } else {
                 connection.sendMessage(buffer.data, buffer.offset, buffer.length)
               }
@@ -622,4 +622,4 @@ object WebSocketTransportFactory extends TransportFactory.Provider with
Log {
   }
 
 
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/activemq-apollo/blob/1e3eadff/apollo-distro/src/main/release/examples/stomp/websocket/js/stomp.js
----------------------------------------------------------------------
diff --git a/apollo-distro/src/main/release/examples/stomp/websocket/js/stomp.js b/apollo-distro/src/main/release/examples/stomp/websocket/js/stomp.js
index 3e237f1..9299345 100644
--- a/apollo-distro/src/main/release/examples/stomp/websocket/js/stomp.js
+++ b/apollo-distro/src/main/release/examples/stomp/websocket/js/stomp.js
@@ -1,16 +1,24 @@
-// Generated by CoffeeScript 1.3.3
+// Generated by CoffeeScript 1.6.3
+/*
+   Stomp Over WebSocket http://www.jmesnil.net/stomp-websocket/doc/ | Apache License V2.0
+
+   Copyright (C) 2010-2013 [Jeff Mesnil](http://jmesnil.net/)
+   Copyright (C) 2012 [FuseSource, Inc.](http://fusesource.com)
+*/
+
+
 (function() {
   var Byte, Client, Frame, Stomp,
-    __hasProp = {}.hasOwnProperty;
-  
-  var MAX_FRAME_SIZE=16*1024;;
-  
+    __hasProp = {}.hasOwnProperty,
+    __slice = [].slice;
+
   Byte = {
     LF: '\x0A',
     NULL: '\x00'
   };
 
   Frame = (function() {
+    var unmarshallSingle;
 
     function Frame(command, headers, body) {
       this.command = command;
@@ -28,14 +36,22 @@
         lines.push("" + name + ":" + value);
       }
       if (this.body) {
-        lines.push("content-length:" + ('' + this.body).length);
+        lines.push("content-length:" + (Frame.sizeOfUTF8(this.body)));
       }
       lines.push(Byte.LF + this.body);
       return lines.join(Byte.LF);
     };
 
-    Frame._unmarshallSingle = function(data) {
-      var body, chr, command, divider, headerLines, headers, i, idx, len, line, start, trim,
_i, _j, _ref, _ref1;
+    Frame.sizeOfUTF8 = function(s) {
+      if (s) {
+        return encodeURI(s).split(/%..|./).length - 1;
+      } else {
+        return 0;
+      }
+    };
+
+    unmarshallSingle = function(data) {
+      var body, chr, command, divider, headerLines, headers, i, idx, len, line, start, trim,
_i, _j, _len, _ref, _ref1;
       divider = data.search(RegExp("" + Byte.LF + Byte.LF));
       headerLines = data.substring(0, divider).split(Byte.LF);
       command = headerLines.shift();
@@ -43,9 +59,9 @@
       trim = function(str) {
         return str.replace(/^\s+|\s+$/g, '');
       };
-      line = idx = null;
-      for (i = _i = 0, _ref = headerLines.length; 0 <= _ref ? _i < _ref : _i > _ref;
i = 0 <= _ref ? ++_i : --_i) {
-        line = headerLines[i];
+      _ref = headerLines.reverse();
+      for (_i = 0, _len = _ref.length; _i < _len; _i++) {
+        line = _ref[_i];
         idx = line.indexOf(':');
         headers[trim(line.substring(0, idx))] = trim(line.substring(idx + 1));
       }
@@ -76,7 +92,7 @@
         for (_i = 0, _len = _ref.length; _i < _len; _i++) {
           data = _ref[_i];
           if ((data != null ? data.length : void 0) > 0) {
-            _results.push(Frame._unmarshallSingle(data));
+            _results.push(unmarshallSingle(data));
           }
         }
         return _results;
@@ -94,6 +110,7 @@
   })();
 
   Client = (function() {
+    var now;
 
     function Client(ws) {
       this.ws = ws;
@@ -104,19 +121,36 @@
         outgoing: 10000,
         incoming: 10000
       };
+      this.maxWebSocketFrameSize = 16 * 1024;
       this.subscriptions = {};
     }
 
+    Client.prototype.debug = function(message) {
+      var _ref;
+      return typeof window !== "undefined" && window !== null ? (_ref = window.console)
!= null ? _ref.log(message) : void 0 : void 0;
+    };
+
+    now = function() {
+      if (Date.now) {
+        return Date.now();
+      } else {
+        return new Date().valueOf;
+      }
+    };
+
     Client.prototype._transmit = function(command, headers, body) {
       var out;
       out = Frame.marshall(command, headers, body);
       if (typeof this.debug === "function") {
         this.debug(">>> " + out);
       }
-      while( true) {
-        if( out.length > MAX_FRAME_SIZE ) {
-          this.ws.send(out.substring(0, MAX_FRAME_SIZE));
-          out = out.substring(MAX_FRAME_SIZE);
+      while (true) {
+        if (out.length > this.maxWebSocketFrameSize) {
+          this.ws.send(out.substring(0, this.maxWebSocketFrameSize));
+          out = out.substring(this.maxWebSocketFrameSize);
+          if (typeof this.debug === "function") {
+            this.debug("remaining = " + out.length);
+          }
         } else {
           return this.ws.send(out);
         }
@@ -144,37 +178,64 @@
         if (typeof this.debug === "function") {
           this.debug("send PING every " + ttl + "ms");
         }
-        this.pinger = typeof window !== "undefined" && window !== null ? window.setInterval(function()
{
+        this.pinger = Stomp.setInterval(ttl, function() {
           _this.ws.send(Byte.LF);
           return typeof _this.debug === "function" ? _this.debug(">>> PING") : void
0;
-        }, ttl) : void 0;
+        });
       }
       if (!(this.heartbeat.incoming === 0 || serverOutgoing === 0)) {
         ttl = Math.max(this.heartbeat.incoming, serverOutgoing);
         if (typeof this.debug === "function") {
           this.debug("check PONG every " + ttl + "ms");
         }
-        return this.ponger = typeof window !== "undefined" && window !== null ? window.setInterval(function()
{
+        return this.ponger = Stomp.setInterval(ttl, function() {
           var delta;
-          delta = Date.now() - _this.serverActivity;
+          delta = now() - _this.serverActivity;
           if (delta > ttl * 2) {
             if (typeof _this.debug === "function") {
               _this.debug("did not receive server activity for the last " + delta + "ms");
             }
-            return _this._cleanUp();
+            return _this.ws.close();
+          }
+        });
+      }
+    };
+
+    Client.prototype._parseConnect = function() {
+      var args, connectCallback, errorCallback, headers;
+      args = 1 <= arguments.length ? __slice.call(arguments, 0) : [];
+      headers = {};
+      switch (args.length) {
+        case 2:
+          headers = args[0], connectCallback = args[1];
+          break;
+        case 3:
+          if (args[1] instanceof Function) {
+            headers = args[0], connectCallback = args[1], errorCallback = args[2];
+          } else {
+            headers.login = args[0], headers.passcode = args[1], connectCallback = args[2];
           }
-        }, ttl) : void 0;
+          break;
+        case 4:
+          headers.login = args[0], headers.passcode = args[1], connectCallback = args[2],
errorCallback = args[3];
+          break;
+        default:
+          headers.login = args[0], headers.passcode = args[1], connectCallback = args[2],
errorCallback = args[3], headers.host = args[4];
       }
+      return [headers, connectCallback, errorCallback];
     };
 
-    Client.prototype.connect = function(login, passcode, connectCallback, errorCallback,
vhost) {
-      var _this = this;
-      this.connectCallback = connectCallback;
+    Client.prototype.connect = function() {
+      var args, errorCallback, headers, out,
+        _this = this;
+      args = 1 <= arguments.length ? __slice.call(arguments, 0) : [];
+      out = this._parseConnect.apply(this, args);
+      headers = out[0], this.connectCallback = out[1], errorCallback = out[2];
       if (typeof this.debug === "function") {
         this.debug("Opening Web Socket...");
       }
       this.ws.onmessage = function(evt) {
-        var arr, c, data, frame, onreceive, _i, _len, _ref, _results;
+        var arr, c, client, data, frame, messageID, onreceive, subscription, _i, _len, _ref,
_results;
         data = typeof ArrayBuffer !== 'undefined' && evt.data instanceof ArrayBuffer
? (arr = new Uint8Array(evt.data), typeof _this.debug === "function" ? _this.debug("--- got
data length: " + arr.length) : void 0, ((function() {
           var _i, _len, _results;
           _results = [];
@@ -184,7 +245,7 @@
           }
           return _results;
         })()).join('')) : evt.data;
-        _this.serverActivity = Date.now();
+        _this.serverActivity = now();
         if (data === Byte.LF) {
           if (typeof _this.debug === "function") {
             _this.debug("<<< PONG");
@@ -208,8 +269,27 @@
               _results.push(typeof _this.connectCallback === "function" ? _this.connectCallback(frame)
: void 0);
               break;
             case "MESSAGE":
-              onreceive = _this.subscriptions[frame.headers.subscription];
-              _results.push(typeof onreceive === "function" ? onreceive(frame) : void 0);
+              subscription = frame.headers.subscription;
+              onreceive = _this.subscriptions[subscription] || _this.onreceive;
+              if (onreceive) {
+                client = _this;
+                messageID = frame.headers["message-id"];
+                frame.ack = function(headers) {
+                  if (headers == null) {
+                    headers = {};
+                  }
+                  return client.ack(messageID, subscription, headers);
+                };
+                frame.nack = function(headers) {
+                  if (headers == null) {
+                    headers = {};
+                  }
+                  return client.nack(messageID, subscription, headers);
+                };
+                _results.push(onreceive(frame));
+              } else {
+                _results.push(typeof _this.debug === "function" ? _this.debug("Unhandled
received MESSAGE: " + frame) : void 0);
+              }
               break;
             case "RECEIPT":
               _results.push(typeof _this.onreceipt === "function" ? _this.onreceipt(frame)
: void 0);
@@ -229,47 +309,37 @@
         if (typeof _this.debug === "function") {
           _this.debug(msg);
         }
+        _this._cleanUp();
         return typeof errorCallback === "function" ? errorCallback(msg) : void 0;
       };
       return this.ws.onopen = function() {
-        var headers;
         if (typeof _this.debug === "function") {
           _this.debug('Web Socket Opened...');
         }
-        headers = {
-          "accept-version": Stomp.VERSIONS.supportedVersions(),
-          "heart-beat": [_this.heartbeat.outgoing, _this.heartbeat.incoming].join(',')
-        };
-        if (vhost) {
-          headers.host = vhost;
-        }
-        if (login) {
-          headers.login = login;
-        }
-        if (passcode) {
-          headers.passcode = passcode;
-        }
+        headers["accept-version"] = Stomp.VERSIONS.supportedVersions();
+        headers["heart-beat"] = [_this.heartbeat.outgoing, _this.heartbeat.incoming].join(',');
         return _this._transmit("CONNECT", headers);
       };
     };
 
-    Client.prototype.disconnect = function(disconnectCallback) {
-      this._transmit("DISCONNECT");
+    Client.prototype.disconnect = function(disconnectCallback, headers) {
+      if (headers == null) {
+        headers = {};
+      }
+      this._transmit("DISCONNECT", headers);
       this.ws.onclose = null;
+      this.ws.close();
       this._cleanUp();
       return typeof disconnectCallback === "function" ? disconnectCallback() : void 0;
     };
 
     Client.prototype._cleanUp = function() {
-      this.ws.close();
       this.connected = false;
       if (this.pinger) {
-        if (typeof window !== "undefined" && window !== null) {
-          window.clearInterval(this.pinger);
-        }
+        Stomp.clearInterval(this.pinger);
       }
       if (this.ponger) {
-        return typeof window !== "undefined" && window !== null ? window.clearInterval(this.ponger)
: void 0;
+        return Stomp.clearInterval(this.ponger);
       }
     };
 
@@ -285,6 +355,7 @@
     };
 
     Client.prototype.subscribe = function(destination, callback, headers) {
+      var client;
       if (headers == null) {
         headers = {};
       }
@@ -294,7 +365,13 @@
       headers.destination = destination;
       this.subscriptions[headers.id] = callback;
       this._transmit("SUBSCRIBE", headers);
-      return headers.id;
+      client = this;
+      return {
+        id: headers.id,
+        unsubscribe: function() {
+          return client.unsubscribe(headers.id);
+        }
+      };
     };
 
     Client.prototype.unsubscribe = function(id) {
@@ -305,9 +382,21 @@
     };
 
     Client.prototype.begin = function(transaction) {
-      return this._transmit("BEGIN", {
-        transaction: transaction
+      var client, txid;
+      txid = transaction || "tx-" + this.counter++;
+      this._transmit("BEGIN", {
+        transaction: txid
       });
+      client = this;
+      return {
+        id: txid,
+        commit: function() {
+          return client.commit(txid);
+        },
+        abort: function() {
+          return client.abort(txid);
+        }
+      };
     };
 
     Client.prototype.commit = function(transaction) {
@@ -345,7 +434,6 @@
   })();
 
   Stomp = {
-    libVersion: "2.0.0-next",
     VERSIONS: {
       V1_0: '1.0',
       V1_1: '1.1',
@@ -370,10 +458,17 @@
   };
 
   if (typeof window !== "undefined" && window !== null) {
+    Stomp.setInterval = function(interval, f) {
+      return window.setInterval(f, interval);
+    };
+    Stomp.clearInterval = function(id) {
+      return window.clearInterval(id);
+    };
     window.Stomp = Stomp;
-  } else {
+  } else if (typeof exports !== "undefined" && exports !== null) {
     exports.Stomp = Stomp;
-    Stomp.WebSocketClass = require('./test/server.mock.js').StompServerMock;
+  } else {
+    self.Stomp = Stomp;
   }
 
 }).call(this);


Mime
View raw message