Author: chirino
Date: Mon Aug 13 16:31:51 2012
New Revision: 1372495
URL: http://svn.apache.org/viewvc?rev=1372495&view=rev
Log:
move the destination_metrics aggregation logic into virtual host so that it can
be tested from unit tests too.
Modified:
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/VirtualHost.scala
activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/BrokerFunSuiteSupport.scala
activemq/activemq-apollo/trunk/apollo-web/src/main/scala/org/apache/activemq/apollo/web/resources/BrokerResource.scala
Modified:
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/VirtualHost.scala
URL:
http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/VirtualHost.scala?rev=1372495&r1=1372494&r2=1372495&view=diff
==============================================================================
---
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/VirtualHost.scala
(original)
+++
activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/VirtualHost.scala
Mon Aug 13 16:31:51 2012
@@ -335,4 +335,24 @@ class VirtualHost(val broker: Broker, va
})
}
+ def get_dest_metrics:FutureResult[AggregateDestMetricsDTO] = {
+ // zero out the enqueue stats on the dsubs since they will already be
accounted for in the topic
+ // stats.
+ Future.all(List(get_queue_metrics, get_topic_metrics,
get_dsub_metrics)).map { x =>
+ val y = x.toArray
+ val (queue, topic, dsub) = (y(0), y(1), y(2))
+ var rc = new AggregateDestMetricsDTO
+ for( queue <- queue.success_option; topic <- topic.success_option; dsub
<- dsub.success_option ) {
+ dsub.enqueue_item_counter = 0L
+ dsub.enqueue_size_counter = 0L
+ dsub.enqueue_ts = 0L
+ rc = aggregate_dest_metrics(List(queue, dsub))
+ DestinationMetricsSupport.add_destination_metrics(rc, topic)
+ rc.objects += topic.objects
+ rc.current_time = broker.now
+ }
+ Success(rc)
+ }
+ }
+
}
Modified:
activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/BrokerFunSuiteSupport.scala
URL:
http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/BrokerFunSuiteSupport.scala?rev=1372495&r1=1372494&r2=1372495&view=diff
==============================================================================
---
activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/BrokerFunSuiteSupport.scala
(original)
+++
activemq/activemq-apollo/trunk/apollo-broker/src/test/scala/BrokerFunSuiteSupport.scala
Mon Aug 13 16:31:51 2012
@@ -21,7 +21,7 @@ import java.net.InetSocketAddress
import org.apache.activemq.apollo.util._
import FileSupport._
import org.fusesource.hawtdispatch._
-import org.apache.activemq.apollo.dto.{AggregateDestMetricsDTO,
QueueStatusDTO, TopicStatusDTO}
+import org.apache.activemq.apollo.dto.{DestMetricsDTO,
AggregateDestMetricsDTO, QueueStatusDTO, TopicStatusDTO}
import collection.immutable.HashMap
import java.io.File
import org.scalatest.{ParallelTestExecution, OneInstancePerTest}
@@ -69,24 +69,19 @@ object BrokerTestSupport {
}
def get_queue_metrics(broker:Broker): AggregateDestMetricsDTO = {
- val host = broker.default_virtual_host
- sync(host) {
- host.get_queue_metrics
- }
+ unwrap_future_result(broker.default_virtual_host.get_queue_metrics)
}
def get_topic_metrics(broker:Broker): AggregateDestMetricsDTO = {
- val host = broker.default_virtual_host
- sync(host) {
- host.get_topic_metrics
- }
+ unwrap_future_result(broker.default_virtual_host.get_topic_metrics)
}
def get_dsub_metrics(broker:Broker): AggregateDestMetricsDTO = {
- val host = broker.default_virtual_host
- sync(host) {
- host.get_dsub_metrics
- }
+ unwrap_future_result(broker.default_virtual_host.get_dsub_metrics)
+ }
+
+ def get_dest_metrics(broker:Broker):AggregateDestMetricsDTO = {
+ unwrap_future_result(broker.default_virtual_host.get_dest_metrics)
}
def queue_status(broker:Broker, name: String): QueueStatusDTO = {
@@ -179,6 +174,7 @@ class BrokerFunSuiteSupport extends FunS
def get_queue_metrics = BrokerTestSupport.get_queue_metrics(broker)
def get_topic_metrics = BrokerTestSupport.get_topic_metrics(broker)
def get_dsub_metrics = BrokerTestSupport.get_dsub_metrics(broker)
+ def get_dest_metrics = BrokerTestSupport.get_dest_metrics(broker)
def queue_status(name: String) = BrokerTestSupport.queue_status(broker, name)
def dsub_status(name: String) = BrokerTestSupport.dsub_status(broker, name)
def webadmin_uri(scheme:String = "http") =
BrokerTestSupport.webadmin_uri(broker, scheme)
Modified:
activemq/activemq-apollo/trunk/apollo-web/src/main/scala/org/apache/activemq/apollo/web/resources/BrokerResource.scala
URL:
http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-web/src/main/scala/org/apache/activemq/apollo/web/resources/BrokerResource.scala?rev=1372495&r1=1372494&r2=1372495&view=diff
==============================================================================
---
activemq/activemq-apollo/trunk/apollo-web/src/main/scala/org/apache/activemq/apollo/web/resources/BrokerResource.scala
(original)
+++
activemq/activemq-apollo/trunk/apollo-web/src/main/scala/org/apache/activemq/apollo/web/resources/BrokerResource.scala
Mon Aug 13 16:31:51 2012
@@ -385,7 +385,11 @@ class BrokerResource() extends Resource
@GET @Path("/virtual-hosts/{id}/dest-metrics")
@ApiOperation(value = "Aggregates the messaging metrics for all the
destinations")
def virtual_host_dest_metrics(@PathParam("id") id : String):
AggregateDestMetricsDTO = {
- aggregate(virtual_host_queue_metrics(id), virtual_host_topic_metrics(id),
virtual_host_dsub_metrics(id))
+ with_virtual_host(id) { host =>
+ monitoring(host) {
+ host.get_dest_metrics
+ }
+ }
}
