activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r1337330 - in /activemq/activemq-apollo/trunk: apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/ apollo-distro/src/main/release/examples/websocket/js/ apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/
Date Fri, 11 May 2012 19:04:02 GMT
Author: chirino
Date: Fri May 11 19:04:02 2012
New Revision: 1337330

URL: http://svn.apache.org/viewvc?rev=1337330&view=rev
Log:
Fixes APLO-199 : Can not consume many messages on Web Sockets protocol

Modified:
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
    activemq/activemq-apollo/trunk/apollo-distro/src/main/release/examples/websocket/js/stomp.js
    activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala?rev=1337330&r1=1337329&r2=1337330&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
(original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/jetty/WebSocketTransportFactory.scala
Fri May 11 19:04:02 2012
@@ -487,23 +487,27 @@ object WebSocketTransportFactory extends
     }
 
     var outbound_capacity_remaining = 1024 * 64;
-    var outbound_drained = 0
 
     val outbound_executor = new SerialExecutor(Broker.BLOCKABLE_THREAD_POOL) {
+      var outbound_drained = 0
       override def drained  = {
-        outbound_capacity_remaining += outbound_drained
+        val amount = outbound_drained
         outbound_drained = 0
+        dispatch_queue {
+          outbound_capacity_remaining += amount
+          transportListener.onRefill()
+        }
       }
     }
 
-    def write(buf: ByteBuffer) = {
+    def write(buf: ByteBuffer):Int = {
       dispatchQueue.assertExecuting
-      var remaining = buf.remaining()
-      if( remaining > 0 ) {
-        if (outbound_capacity_remaining <= 0) {
-          outbound_capacity_remaining -= remaining;
-        }
+      val remaining = buf.remaining()
+      if( remaining==0 )
+        return 0
 
+      if( outbound_capacity_remaining > 0 ) {
+        outbound_capacity_remaining -= remaining;
         var buffer = new Buffer(buf.array(), buf.arrayOffset(), buf.remaining())
         outbound_executor {
           if( !binary_transfers ) {
@@ -511,11 +515,14 @@ object WebSocketTransportFactory extends
           } else {
             connection.sendMessage(buffer.data, buffer.offset, buffer.length)
           }
-          outbound_drained += remaining
+          outbound_executor.outbound_drained += remaining
         }
         buf.position(buf.position()+ remaining);
+        return remaining
+
+      } else {
+        return 0
       }
-      remaining;
     }
 
   }

Modified: activemq/activemq-apollo/trunk/apollo-distro/src/main/release/examples/websocket/js/stomp.js
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-distro/src/main/release/examples/websocket/js/stomp.js?rev=1337330&r1=1337329&r2=1337330&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-distro/src/main/release/examples/websocket/js/stomp.js
(original)
+++ activemq/activemq-apollo/trunk/apollo-distro/src/main/release/examples/websocket/js/stomp.js
Fri May 11 19:04:02 2012
@@ -1,17 +1,23 @@
+// Generated by CoffeeScript 1.3.1
 
 /*
 Copyright (C) 2010 Jeff Mesnil -- http://jmesnil.net/
 Copyright (C) 2012 FuseSource, Inc. -- http://fusesource.com
 */
 
+
 (function() {
   var Client, Stomp, WebSocketStompMock,
-    __hasProp = Object.prototype.hasOwnProperty;
+    __hasProp = {}.hasOwnProperty;
 
   Stomp = {
     frame: function(command, headers, body) {
-      if (headers == null) headers = [];
-      if (body == null) body = '';
+      if (headers == null) {
+        headers = [];
+      }
+      if (body == null) {
+        body = '';
+      }
       return {
         command: command,
         headers: headers,
@@ -36,7 +42,7 @@ Copyright (C) 2012 FuseSource, Inc. -- h
       };
     },
     unmarshal: function(data) {
-      var body, chr, command, divider, headerLines, headers, i, idx, line, trim, _ref, _ref2,
_ref3;
+      var body, chr, command, divider, headerLines, headers, i, idx, line, trim, _i, _j,
_ref, _ref1, _ref2;
       divider = data.search(/\n\n/);
       headerLines = data.substring(0, divider).split('\n');
       command = headerLines.shift();
@@ -46,21 +52,39 @@ Copyright (C) 2012 FuseSource, Inc. -- h
         return str.replace(/^\s+/g, '').replace(/\s+$/g, '');
       };
       line = idx = null;
-      for (i = 0, _ref = headerLines.length; 0 <= _ref ? i < _ref : i > _ref; 0
<= _ref ? i++ : i--) {
+      for (i = _i = 0, _ref = headerLines.length; 0 <= _ref ? _i < _ref : _i > _ref;
i = 0 <= _ref ? ++_i : --_i) {
         line = headerLines[i];
         idx = line.indexOf(':');
         headers[trim(line.substring(0, idx))] = trim(line.substring(idx + 1));
       }
       chr = null;
-      for (i = _ref2 = divider + 2, _ref3 = data.length; _ref2 <= _ref3 ? i < _ref3
: i > _ref3; _ref2 <= _ref3 ? i++ : i--) {
+      for (i = _j = _ref1 = divider + 2, _ref2 = data.length; _ref1 <= _ref2 ? _j <
_ref2 : _j > _ref2; i = _ref1 <= _ref2 ? ++_j : --_j) {
         chr = data.charAt(i);
-        if (chr === '\0') break;
+        if (chr === '\x00') {
+          break;
+        }
         body += chr;
       }
       return Stomp.frame(command, headers, body);
     },
+    unmarshal_multi: function(multi_datas) {
+      var data, datas;
+      datas = (function() {
+        var _i, _len, _ref, _results;
+        _ref = multi_datas.split(/\x00\n*/);
+        _results = [];
+        for (_i = 0, _len = _ref.length; _i < _len; _i++) {
+          data = _ref[_i];
+          if (data && data.length > 0) {
+            _results.push(Stomp.unmarshal(data));
+          }
+        }
+        return _results;
+      })();
+      return datas;
+    },
     marshal: function(command, headers, body) {
-      return Stomp.frame(command, headers, body).toString() + '\0';
+      return Stomp.frame(command, headers, body).toString() + '\x00';
     },
     client: function(url) {
       return new Client(url);
@@ -69,6 +93,8 @@ Copyright (C) 2012 FuseSource, Inc. -- h
 
   Client = (function() {
 
+    Client.name = 'Client';
+
     function Client(url) {
       this.url = url;
       this.counter = 0;
@@ -79,19 +105,23 @@ Copyright (C) 2012 FuseSource, Inc. -- h
     Client.prototype._transmit = function(command, headers, body) {
       var out;
       out = Stomp.marshal(command, headers, body);
-      if (typeof this.debug === "function") this.debug(">>> " + out);
+      if (typeof this.debug === "function") {
+        this.debug(">>> " + out);
+      }
       return this.ws.send(out);
     };
 
     Client.prototype.connect = function(login_, passcode_, connectCallback, errorCallback)
{
       var klass,
         _this = this;
-      if (typeof this.debug === "function") this.debug("Opening Web Socket...");
+      if (typeof this.debug === "function") {
+        this.debug("Opening Web Socket...");
+      }
       klass = WebSocketStompMock || WebSocket;
       this.ws = new klass(this.url);
       this.ws.binaryType = "arraybuffer";
       this.ws.onmessage = function(evt) {
-        var data, frame, i, onreceive, view;
+        var data, frame, i, onreceive, view, _i, _len, _ref, _results;
         data = (function() {
           var _i, _len;
           if (evt.data instanceof ArrayBuffer) {
@@ -109,24 +139,39 @@ Copyright (C) 2012 FuseSource, Inc. -- h
             return evt.data;
           }
         }).call(_this);
-        if (typeof _this.debug === "function") _this.debug('<<< ' + data);
-        frame = Stomp.unmarshal(data);
-        if (frame.command === "CONNECTED" && connectCallback) {
-          _this.connected = true;
-          return connectCallback(frame);
-        } else if (frame.command === "MESSAGE") {
-          onreceive = _this.subscriptions[frame.headers.subscription];
-          return typeof onreceive === "function" ? onreceive(frame) : void 0;
+        if (typeof _this.debug === "function") {
+          _this.debug('<<< ' + data);
         }
+        _ref = Stomp.unmarshal_multi(data);
+        _results = [];
+        for (_i = 0, _len = _ref.length; _i < _len; _i++) {
+          frame = _ref[_i];
+          if (frame.command === "CONNECTED" && connectCallback) {
+            _this.connected = true;
+            _results.push(connectCallback(frame));
+          } else if (frame.command === "MESSAGE") {
+            onreceive = _this.subscriptions[frame.headers.subscription];
+            _results.push(typeof onreceive === "function" ? onreceive(frame) : void 0);
+          } else if (frame.command === "ERROR") {
+            _results.push(typeof errorCallback === "function" ? errorCallback(frame) : void
0);
+          } else {
+            _results.push(typeof _this.debug === "function" ? _this.debug("Unhandled frame:
" + frame) : void 0);
+          }
+        }
+        return _results;
       };
       this.ws.onclose = function() {
         var msg;
         msg = "Whoops! Lost connection to " + _this.url;
-        if (typeof _this.debug === "function") _this.debug(msg);
+        if (typeof _this.debug === "function") {
+          _this.debug(msg);
+        }
         return typeof errorCallback === "function" ? errorCallback(msg) : void 0;
       };
       this.ws.onopen = function() {
-        if (typeof _this.debug === "function") _this.debug('Web Socket Opened...');
+        if (typeof _this.debug === "function") {
+          _this.debug('Web Socket Opened...');
+        }
         return _this._transmit("CONNECT", {
           login: login_,
           passcode: passcode_
@@ -143,15 +188,21 @@ Copyright (C) 2012 FuseSource, Inc. -- h
     };
 
     Client.prototype.send = function(destination, headers, body) {
-      if (headers == null) headers = {};
-      if (body == null) body = '';
+      if (headers == null) {
+        headers = {};
+      }
+      if (body == null) {
+        body = '';
+      }
       headers.destination = destination;
       return this._transmit("SEND", headers, body);
     };
 
     Client.prototype.subscribe = function(destination, callback, headers) {
       var id;
-      if (headers == null) headers = {};
+      if (headers == null) {
+        headers = {};
+      }
       id = "sub-" + this.counter++;
       headers.destination = destination;
       headers.id = id;
@@ -161,32 +212,42 @@ Copyright (C) 2012 FuseSource, Inc. -- h
     };
 
     Client.prototype.unsubscribe = function(id, headers) {
-      if (headers == null) headers = {};
+      if (headers == null) {
+        headers = {};
+      }
       headers.id = id;
       delete this.subscriptions[id];
       return this._transmit("UNSUBSCRIBE", headers);
     };
 
     Client.prototype.begin = function(transaction, headers) {
-      if (headers == null) headers = {};
+      if (headers == null) {
+        headers = {};
+      }
       headers.transaction = transaction;
       return this._transmit("BEGIN", headers);
     };
 
     Client.prototype.commit = function(transaction, headers) {
-      if (headers == null) headers = {};
+      if (headers == null) {
+        headers = {};
+      }
       headers.transaction = transaction;
       return this._transmit("COMMIT", headers);
     };
 
     Client.prototype.abort = function(transaction, headers) {
-      if (headers == null) headers = {};
+      if (headers == null) {
+        headers = {};
+      }
       headers.transaction = transaction;
       return this._transmit("ABORT", headers);
     };
 
     Client.prototype.ack = function(message_id, headers) {
-      if (headers == null) headers = {};
+      if (headers == null) {
+        headers = {};
+      }
       headers["message-id"] = message_id;
       return this._transmit("ACK", headers);
     };

Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala?rev=1337330&r1=1337329&r2=1337330&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
(original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
Fri May 11 19:04:02 2012
@@ -390,7 +390,10 @@ class StompProtocolHandler extends Proto
 
     val consumer_sink = sink_manager.open()
     val credit_window_filter = new CreditWindowFilter[Delivery](consumer_sink.map { delivery
=>
-      ack_handler.track(delivery)
+
+      if( !dead ) {
+        ack_handler.track(delivery)
+      }
 
       val message = delivery.message
       var frame = if( message.protocol eq StompProtocol ) {



Mime
View raw message