You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2012/05/11 21:04:02 UTC
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/
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 ) {