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 2010/11/29 14:48:31 UTC
svn commit: r1040114 - in /activemq/activemq-apollo/trunk:
apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/
apollo-broker/src/test/scala/org/apache/activemq/apollo/broker/perf/
apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp...
Author: chirino
Date: Mon Nov 29 13:48:30 2010
New Revision: 1040114
URL: http://svn.apache.org/viewvc?rev=1040114&view=rev
Log:
Last commit impacted perf due to Destination case class not properly implementing equals. Fixed.
Removed:
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Destination.java
Modified:
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Destinations.scala
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Router.scala
activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/org/apache/activemq/apollo/broker/perf/BrokerPerfSupport.scala
activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala
activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocol.scala
activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompRemoteClients.scala
Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Destinations.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Destinations.scala?rev=1040114&r1=1040113&r2=1040114&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Destinations.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Destinations.scala Mon Nov 29 13:48:30 2010
@@ -22,6 +22,14 @@ import org.apache.activemq.apollo.util.p
import Buffer._
/**
+ */
+trait Destination {
+ def domain:AsciiBuffer = null
+ def name:Array[Path] = null
+ def destinations:List[Destination] = null
+}
+
+/**
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
class DestinationParser extends PathParser {
@@ -39,17 +47,17 @@ class DestinationParser extends PathPars
} else {
val baos = new ByteArrayOutputStream
def write(value: Destination):Unit = {
- if (value.getDestinations != null) {
+ if (value.destinations != null) {
assert( destination_separator.isDefined )
val first = true
- for (d <- value.getDestinations) {
+ for (d <- value.destinations) {
if (!first) {
baos.write(destination_separator.get)
}
write(d)
}
} else {
- value.getDomain match {
+ value.domain match {
case Router.QUEUE_DOMAIN =>
baos.write(queue_prefix)
case Router.TOPIC_DOMAIN =>
@@ -59,7 +67,7 @@ class DestinationParser extends PathPars
case Router.TEMP_TOPIC_DOMAIN =>
baos.write(temp_topic_prefix)
}
- this.write(value.path, baos)
+ this.write(value.name.toArray, baos)
}
}
write(value)
@@ -89,25 +97,25 @@ class DestinationParser extends PathPars
}
dl = dl ::: d :: Nil
}
- return new MultiDestination(dl.toArray[Destination]);
+ return new MultiDestination(dl);
} else {
if (queue_prefix != null && value.startsWith(queue_prefix)) {
var name = value.slice(queue_prefix.length, value.length).ascii();
- return new SingleDestination(Router.QUEUE_DOMAIN, parsePath(name));
+ return new SingleDestination(Router.QUEUE_DOMAIN, parsePath(name).toList);
} else if (topic_prefix != null && value.startsWith(topic_prefix)) {
var name = value.slice(topic_prefix.length, value.length).ascii();
- return new SingleDestination(Router.TOPIC_DOMAIN, parsePath(name));
+ return new SingleDestination(Router.TOPIC_DOMAIN, parsePath(name).toList);
} else if (temp_queue_prefix != null && value.startsWith(temp_queue_prefix)) {
var name = value.slice(temp_queue_prefix.length, value.length).ascii();
- return new SingleDestination(Router.TEMP_QUEUE_DOMAIN, parsePath(name));
+ return new SingleDestination(Router.TEMP_QUEUE_DOMAIN, parsePath(name).toList);
} else if (temp_topic_prefix != null && value.startsWith(temp_topic_prefix)) {
var name = value.slice(temp_topic_prefix.length, value.length).ascii();
- return new SingleDestination(Router.TEMP_TOPIC_DOMAIN, parsePath(name));
+ return new SingleDestination(Router.TEMP_TOPIC_DOMAIN, parsePath(name).toList);
} else {
if (default_domain == null) {
return null;
}
- return new SingleDestination(default_domain, parsePath(value));
+ return new SingleDestination(default_domain, parsePath(value).toList);
}
}
}
@@ -116,23 +124,14 @@ class DestinationParser extends PathPars
/**
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
-case class SingleDestination(var domain: AsciiBuffer = null, var name: Array[Path] = null) extends Destination {
- def getDestinations(): Array[Destination] = null;
- def getDomain(): AsciiBuffer = domain
-
- def path() = name
-
- override def toString() = "" + domain + ":" + name
+case class SingleDestination(override val domain: AsciiBuffer, val path: List[Path]) extends Destination {
+ override def toString() = "" + domain + ":" + name.mkString(".")
+ override val name = path.toArray[Path]
}
/**
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
-case class MultiDestination(var destinations: Array[Destination]) extends Destination {
- def getDestinations(): Array[Destination] = destinations;
- def getDomain(): AsciiBuffer = null
-
- def path() = null
-
+case class MultiDestination(override val destinations: List[Destination]) extends Destination {
override def toString() = destinations.mkString(",")
}
Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Router.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Router.scala?rev=1040114&r1=1040113&r2=1040114&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Router.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Router.scala Mon Nov 29 13:48:30 2010
@@ -77,7 +77,7 @@ class Router(val host:VirtualHost) exten
val bindings = new PathMap[Queue]()
private def is_topic(destination:Destination) = {
- destination.getDomain match {
+ destination.domain match {
case TOPIC_DOMAIN => true
case TEMP_TOPIC_DOMAIN => true
case _ => false
@@ -204,7 +204,7 @@ class Router(val host:VirtualHost) exten
assert( is_topic(destination) )
- val name = destination.path
+ val name = destination.name
// make sure the destination is created if this is not a wild card sub
if( !PathParser.containsWildCards(name) ) {
@@ -220,7 +220,7 @@ class Router(val host:VirtualHost) exten
def unbind(destination:Destination, consumer:DeliveryConsumer) = releasing(consumer) {
assert( is_topic(destination) )
- val name = destination.path
+ val name = destination.name
broadcast_consumers.remove(name, consumer)
get_destination_matches(name).foreach{ node=>
node.remove_broadcast_consumer(consumer)
@@ -243,13 +243,13 @@ class Router(val host:VirtualHost) exten
// Looking up the queue will cause it to get created if it does not exist.
val queue = if( !topic ) {
val dto = new PointToPointBindingDTO
- dto.destination = Binding.encode(destination.path)
+ dto.destination = Binding.encode(destination.name)
_create_queue(dto)
} else {
None
}
- val node = create_destination_or(destination.path) { node=> Unit }
+ val node = create_destination_or(destination.name) { node=> Unit }
if( node.unified || topic ) {
node.add_broadcast_producer( route )
} else {
@@ -263,7 +263,7 @@ class Router(val host:VirtualHost) exten
def disconnect(route:DeliveryProducerRoute) = releasing(route) {
val topic = is_topic(route.destination)
- val node = create_destination_or(route.destination.path) { node=> Unit }
+ val node = create_destination_or(route.destination.name) { node=> Unit }
if( node.unified || topic ) {
node.remove_broadcast_producer(route)
}
Modified: activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/org/apache/activemq/apollo/broker/perf/BrokerPerfSupport.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/org/apache/activemq/apollo/broker/perf/BrokerPerfSupport.scala?rev=1040114&r1=1040113&r2=1040114&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/org/apache/activemq/apollo/broker/perf/BrokerPerfSupport.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/org/apache/activemq/apollo/broker/perf/BrokerPerfSupport.scala Mon Nov 29 13:48:30 2010
@@ -216,7 +216,7 @@ abstract class BrokerPerfSupport extends
for (i <- 0 until destCount) {
val domain = if (PTP) {Router.QUEUE_DOMAIN} else {Router.TOPIC_DOMAIN}
val name = new AsciiBuffer("dest" + (i + 1))
- var bean = new SingleDestination(domain, parser.parsePath(name))
+ var bean = new SingleDestination(domain, parser.parsePath(name).toList)
dests(i) = bean
// if (PTP) {
// sendBroker.defaultVirtualHost.createQueue(dests(i))
Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala?rev=1040114&r1=1040113&r2=1040114&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompFrame.scala Mon Nov 29 13:48:30 2010
@@ -308,7 +308,7 @@ object Stomp {
val destination_parser = new DestinationParser
destination_parser.queue_prefix = ascii("/queue/")
destination_parser.topic_prefix = ascii("/topic/")
- destination_parser.path_seperator = ascii("/")
+ destination_parser.path_seperator = ascii(".")
destination_parser.any_child_wildcard = ascii("*")
destination_parser.any_descendant_wildcard = ascii("**")
Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocol.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocol.scala?rev=1040114&r1=1040113&r2=1040114&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocol.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocol.scala Mon Nov 29 13:48:30 2010
@@ -703,7 +703,7 @@ class StompProtocolHandler extends Proto
}
- val topic = destination.getDomain == Router.TOPIC_DOMAIN
+ val topic = destination.domain == Router.TOPIC_DOMAIN
var persistent = get(headers, PERSISTENT).map( _ == TRUE ).getOrElse(false)
val ack = get(headers, ACK_MODE) match {
@@ -743,7 +743,7 @@ class StompProtocolHandler extends Proto
// way again)
if (topic) {
val rc = new DurableSubscriptionBindingDTO
- rc.destination = Binding.encode(destination.path)
+ rc.destination = Binding.encode(destination.name)
// TODO:
// rc.client_id =
rc.subscription_id = if( persistent ) id else null
@@ -751,7 +751,7 @@ class StompProtocolHandler extends Proto
rc
} else {
val rc = new PointToPointBindingDTO
- rc.destination = Binding.encode(destination.path)
+ rc.destination = Binding.encode(destination.name)
rc
}
}
Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompRemoteClients.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompRemoteClients.scala?rev=1040114&r1=1040113&r2=1040114&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompRemoteClients.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/test/scala/org/apache/activemq/apollo/stomp/perf/StompRemoteClients.scala Mon Nov 29 13:48:30 2010
@@ -38,10 +38,10 @@ class StompRemoteConsumer extends Remote
outboundSink = new OverflowSink[StompFrame](MapSink(transportSink) {x => x})
outboundSink.refiller = ^ {}
- val stompDestination = if (destination.getDomain() == Router.QUEUE_DOMAIN) {
- ascii("/queue/" + destination.path().toString());
+ val stompDestination = if (destination.domain == Router.QUEUE_DOMAIN) {
+ ascii("/queue/" + destination.name.toString());
} else {
- ascii("/topic/" + destination.path().toString());
+ ascii("/topic/" + destination.name.toString());
}
var frame = StompFrame(CONNECT);
@@ -149,10 +149,10 @@ class StompRemoteProducer extends Remote
outboundSink = new OverflowSink[StompFrame](MapSink(transportSink) {x => x})
outboundSink.refiller = ^ {drain}
- if (destination.getDomain() == Router.QUEUE_DOMAIN) {
- stompDestination = ascii("/queue/" + destination.path().toString());
+ if (destination.domain == Router.QUEUE_DOMAIN) {
+ stompDestination = ascii("/queue/" + destination.name.toString());
} else {
- stompDestination = ascii("/topic/" + destination.path().toString());
+ stompDestination = ascii("/topic/" + destination.name.toString());
}
outboundSink.offer(StompFrame(CONNECT));
send_next