activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r1161969 - in /activemq/activemq-apollo/trunk: apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/ apollo-dto/src/main/java/org/apache/activemq/apollo/dto/ apollo-dto/src/main/resources/org/apache/activemq/apollo/dto/ apollo-ope...
Date Fri, 26 Aug 2011 02:49:46 GMT
Author: chirino
Date: Fri Aug 26 02:49:45 2011
New Revision: 1161969

URL: http://svn.apache.org/viewvc?rev=1161969&view=rev
Log:
Fixes https://issues.apache.org/jira/browse/APLO-78, https://issues.apache.org/jira/browse/APLO-80, and https://issues.apache.org/jira/browse/APLO-81 : Improvements and new features for the REST API

Topics now provide messaging metrics similar to the what was provided by queues.  You now also have aggregate statistics are now accessible via the following routes:
/queue-metrics
/topic-metrics
/dsub-metrics
/dest-metrics

dest-metrics aggregates the queue, topic, and dsub metrics.  queue-metrics no longer includes metrics from dsubs since they can be access via /dsub-mtrics.  The JSON structure of the queue and topic json objects have slightly changes to make them more consistent with each other and also provide messaging metrics per producer/consumer.

Added:
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateTopicMetricsDTO.java
      - copied, changed from r1161956, activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/DestinationMetricsDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerLinkDTO.java
      - copied, changed from r1161956, activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerStatusDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicMetricsDTO.java
      - copied, changed from r1161956, activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java
Removed:
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerStatusDTO.java
Modified:
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Broker.scala
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Delivery.scala
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/LocalRouter.scala
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Router.scala
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Topic.scala
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/LinkDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueMetricsDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueStatusDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicStatusDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/VirtualHostStatusDTO.java
    activemq/activemq-apollo/trunk/apollo-dto/src/main/resources/org/apache/activemq/apollo/dto/jaxb.index
    activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
    activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
    activemq/activemq-apollo/trunk/apollo-web/src/main/scala/org/apache/activemq/apollo/web/resources/BrokerResource.scala
    activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/QueueStatusDTO.jade
    activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/TopicStatusDTO.jade

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Broker.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Broker.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Broker.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Broker.scala Fri Aug 26 02:49:45 2011
@@ -34,6 +34,8 @@ import management.ManagementFactory
 import org.apache.activemq.apollo.dto._
 import javax.management.ObjectName
 import org.fusesource.hawtdispatch.TaskTracker._
+import java.util.concurrent.TimeUnit
+import collection.mutable.ListBuffer._
 
 /**
  * <p>
@@ -250,6 +252,9 @@ class Broker() extends BaseService {
 
   var web_server:WebServer = _
 
+  @volatile
+  var now = System.currentTimeMillis()
+
   var config_log:Log = Log(new MemoryLogger(Broker.log))
   var audit_log:Log = Broker
   var security_log:Log  = Broker
@@ -292,7 +297,9 @@ class Broker() extends BaseService {
     log_versions
     check_file_limit
     init_dispatch_queue(dispatch_queue)
+
     BrokerRegistry.add(this)
+    schedule_periodic_maintenance
 
     val tracker = new LoggingTracker("broker startup", console_log, SERVICE_TIMEOUT)
     apply_update(tracker)
@@ -333,6 +340,14 @@ class Broker() extends BaseService {
 
     BrokerRegistry.remove(this)
     tracker.callback(on_completed)
+
+  }
+
+  def schedule_periodic_maintenance:Unit = dispatch_queue.after(1, TimeUnit.SECONDS) {
+    if( service_state.is_started ) {
+      now = System.currentTimeMillis
+      schedule_periodic_maintenance
+    }
   }
 
   protected def init_logs = {

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Delivery.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Delivery.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Delivery.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Delivery.scala Fri Aug 26 02:49:45 2011
@@ -76,6 +76,14 @@ trait DeliveryConsumer extends Retained 
  * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
  */
 trait DeliverySession extends Sink[Delivery] {
+  /**
+   * The number of deliveries accepted by this session.
+   */
+  def enqueue_item_counter:Long
+  /**
+   * The total size of the deliveries accepted by this session.
+   */
+  def enqueue_size_counter:Long
   def producer:DeliveryProducer
   def consumer:DeliveryConsumer
   def remaining_capacity:Int

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/LocalRouter.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/LocalRouter.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/LocalRouter.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/LocalRouter.scala Fri Aug 26 02:49:45 2011
@@ -957,8 +957,8 @@ class LocalRouter(val virtual_host:Virtu
     // For the topics, just collocate the producers onto the first consumer's thread.
     topic_domain.destinations.foreach { node =>
 
-      node.consumers.headOption.foreach{ consumer =>
-        node.producers.foreach { r=>
+      node.consumers.keys.headOption.foreach{ consumer =>
+        node.producers.keys.foreach { r=>
           r.collocate(consumer.dispatch_queue)
         }
       }

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala Fri Aug 26 02:49:45 2011
@@ -27,10 +27,10 @@ import org.apache.activemq.apollo.util.l
 import org.fusesource.hawtdispatch.{ListEventAggregator, DispatchQueue, BaseRetained}
 import OptionSupport._
 import security.SecurityContext
-import org.apache.activemq.apollo.dto.{DestinationDTO, QueueDTO}
 import java.util.concurrent.atomic.{AtomicReference, AtomicLong, AtomicInteger}
 import org.fusesource.hawtbuf.Buffer
 import java.lang.UnsupportedOperationException
+import org.apache.activemq.apollo.dto._
 
 object Queue extends Log {
   val subcsription_counter = new AtomicInteger(0)
@@ -206,6 +206,129 @@ class Queue(val router: LocalRouter, val
   var auto_delete_after = 0
   var idled_at = 0L
 
+  def get_queue_metrics:QueueMetricsDTO = {
+    val rc = new QueueMetricsDTO
+
+    rc.enqueue_item_counter = this.enqueue_item_counter
+    rc.enqueue_size_counter = this.enqueue_size_counter
+    rc.enqueue_ts = this.enqueue_ts
+
+    rc.dequeue_item_counter = this.dequeue_item_counter
+    rc.dequeue_size_counter = this.dequeue_size_counter
+    rc.dequeue_ts = this.dequeue_ts
+
+    rc.nack_item_counter = this.nack_item_counter
+    rc.nack_size_counter = this.nack_size_counter
+    rc.nack_ts = this.nack_ts
+
+    rc.expired_item_counter = this.expired_item_counter
+    rc.expired_size_counter = this.expired_size_counter
+    rc.expired_ts = this.expired_ts
+
+    rc.queue_size = this.queue_size
+    rc.queue_items = this.queue_items
+
+    rc.swap_out_item_counter = this.swap_out_item_counter
+    rc.swap_out_size_counter = this.swap_out_size_counter
+    rc.swap_in_item_counter = this.swap_in_item_counter
+    rc.swap_in_size_counter = this.swap_in_size_counter
+
+    rc.swapping_in_size = this.swapping_in_size
+    rc.swapping_out_size = this.swapping_out_size
+
+    rc.swapped_in_items = this.swapped_in_items
+    rc.swapped_in_size = this.swapped_in_size
+
+    rc.swapped_in_size_max = this.swapped_in_size_max
+
+    rc.producer_counter = this.producer_counter
+    rc.consumer_counter = this.consumer_counter
+
+    rc.producer_count = this.producers.size
+    rc.consumer_count = this.all_subscriptions.size
+    rc
+  }
+
+  def status(entries:Boolean=false) = {
+    val rc = new QueueStatusDTO
+    rc.id = this.id
+    rc.state = this.service_state.toString
+    rc.state_since = this.service_state.since
+    rc.binding = this.binding.binding_dto
+    rc.config = this.config
+    rc.metrics = this.get_queue_metrics
+    rc.metrics.current_time = System.currentTimeMillis()
+
+    if( entries ) {
+      var cur = this.head_entry
+      while( cur!=null ) {
+
+        val e = new EntryStatusDTO
+        e.seq = cur.seq
+        e.count = cur.count
+        e.size = cur.size
+        e.consumer_count = cur.parked.size
+        e.is_prefetched = cur.is_prefetched
+        e.state = cur.label
+
+        rc.entries.add(e)
+
+        cur = if( cur == this.tail_entry ) {
+          null
+        } else {
+          cur.nextOrTail
+        }
+      }
+    }
+
+    this.inbound_sessions.foreach { session:DeliverySession =>
+      val link = new LinkDTO()
+      session.producer.connection match {
+        case Some(connection) =>
+          link.kind = "connection"
+          link.id = connection.id.toString
+          link.label = connection.transport.getRemoteAddress.toString
+        case _ =>
+          link.kind = "unknown"
+          link.label = "unknown"
+      }
+      link.enqueue_item_counter = session.enqueue_item_counter
+      link.enqueue_size_counter = session.enqueue_size_counter
+      rc.producers.add(link)
+    }
+
+    this.all_subscriptions.valuesIterator.toSeq.foreach{ sub =>
+      val link = new QueueConsumerLinkDTO
+      sub.consumer.connection match {
+        case Some(connection) =>
+          link.kind = "connection"
+          link.id = connection.id.toString
+          link.label = connection.transport.getRemoteAddress.toString
+        case _ =>
+          link.kind = "unknown"
+          link.label = "unknown"
+      }
+      link.position = sub.pos.seq
+      link.enqueue_item_counter = sub.total_dispatched_count
+      link.enqueue_size_counter = sub.total_dispatched_size
+      link.total_ack_count = sub.total_ack_count
+      link.total_nack_count = sub.total_nack_count
+      link.acquired_size = sub.acquired_size
+      link.acquired_count = sub.acquired_count
+      link.waiting_on = if( sub.full ) {
+        "ack"
+      } else if( sub.pos.is_tail ) {
+        "producer"
+      } else if( !sub.pos.is_loaded ) {
+        "load"
+      } else {
+        "dispatch"
+      }
+      rc.consumers.add(link)
+    }
+    rc
+  }
+
   def update(on_completed:Runnable) = dispatch_queue {
 
     val prev_persistent = tune_persistent
@@ -630,6 +753,8 @@ class Queue(val router: LocalRouter, val
     }
 
     def remaining_capacity = session.remaining_capacity
+    def enqueue_item_counter = session.accepted_count
+    def enqueue_size_counter = session.accepted_size
 
     def close = {
       session_manager.close(session)

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=1161969&r1=1161968&r2=1161969&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 Fri Aug 26 02:49:45 2011
@@ -91,13 +91,15 @@ abstract class DeliveryProducerRoute(rou
     dispatch_queue {
       consumers.foreach{ x=>
         debug("producer route attaching to conusmer.")
-        val target = x.connect(this);
+        val target = connect(x);
         target.refiller = drainer
         targets ::= target
       }
     }
   }
 
+  def connect(x:DeliveryConsumer) = x.connect(this)
+
   def unbind(targets:List[DeliveryConsumer]) = dispatch_queue {
     this.targets = this.targets.filterNot { x=>
       val rc = targets.contains(x.consumer)
@@ -181,6 +183,7 @@ abstract class DeliveryProducerRoute(rou
     }
   }
 
+
   private def delivered(delivery: Delivery): Unit = {
     if (pendingAck != null) {
       if (delivery.uow != null) {

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala Fri Aug 26 02:49:45 2011
@@ -21,6 +21,7 @@ import org.fusesource.hawtdispatch._
 import java.util.LinkedList
 import org.apache.activemq.apollo.transport.Transport
 import collection.mutable.HashSet
+import java.util.concurrent.atomic.AtomicLong
 
 /**
  * <p>
@@ -229,6 +230,8 @@ class CreditWindowFilter[T](val downstre
 }
 
 trait SessionSink[T] extends Sink[T] {
+  def accepted_count:Long
+  def accepted_size:Long
   def remaining_capacity:Int
 }
 
@@ -322,6 +325,11 @@ class Session[T](val producer_queue:Disp
   private def sizer = mux.sizer
   private def downstream = mux.source
 
+  @volatile
+  var accepted_count = 0L
+  @volatile
+  var accepted_size = 0L
+
   // create a source to coalesce credit events back to the producer side...
   val credit_adder = createSource(EventAggregators.INTEGER_ADD , producer_queue)
   credit_adder.onEvent{
@@ -361,7 +369,12 @@ class Session[T](val producer_queue:Disp
     if( _full || closed ) {
       false
     } else {
-      add_credits(-sizer.size(value))
+      val size = sizer.size(value)
+
+      accepted_count += 1
+      accepted_size += size
+
+      add_credits(-size)
       downstream.merge((this, value))
       true
     }

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Topic.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Topic.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Topic.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Topic.scala Fri Aug 26 02:49:45 2011
@@ -20,9 +20,10 @@ import org.apache.activemq.apollo.util._
 import path.Path
 import scala.collection.immutable.List
 import org.apache.activemq.apollo.dto._
-import collection.mutable.{HashMap, ListBuffer}
 import java.util.concurrent.TimeUnit
 import org.fusesource.hawtdispatch._
+import collection.mutable.{HashSet, HashMap, ListBuffer}
+import java.lang.Long
 
 /**
  * <p>
@@ -33,8 +34,82 @@ import org.fusesource.hawtdispatch._
  */
 class Topic(val router:LocalRouter, val destination_dto:TopicDestinationDTO, var config_updater: ()=>TopicDTO, val id:String, path:Path) extends DomainDestination {
 
-  var producers = ListBuffer[BindableDeliveryProducer]()
-  var consumers = ListBuffer[DeliveryConsumer]()
+  var enqueue_item_counter = 0L
+  var enqueue_size_counter = 0L
+  var enqueue_ts = System.currentTimeMillis()
+
+  var dequeue_item_counter = 0L
+  var dequeue_size_counter = 0L
+  var dequeue_ts = System.currentTimeMillis()
+
+  var proxy_sessions = new HashSet[ProxyDeliverySession]()
+
+  implicit def from_link(from:LinkDTO):(Long,Long,Long)=(from.enqueue_item_counter, from.enqueue_size_counter, from.enqueue_ts)
+  implicit def from_session(from:ProxyDeliverySession):(Long,Long,Long)=(from.enqueue_item_counter, from.enqueue_size_counter, from.enqueue_ts)
+
+  def add_counters(to:LinkDTO, from:(Long,Long,Long)):Unit = {
+    to.enqueue_item_counter += from._1
+    to.enqueue_size_counter += from._2
+    to.enqueue_ts = to.enqueue_ts max from._2
+  }
+  def add_counters(to:Topic, from:(Long,Long,Long)):Unit = {
+    to.enqueue_item_counter += from._1
+    to.enqueue_size_counter += from._2
+    to.enqueue_ts = to.enqueue_ts max from._2
+  }
+
+  case class ProxyDeliverySession(session:DeliverySession) extends DeliverySession with SinkFilter[Delivery] {
+    dispatch_queue {
+      proxy_sessions.add(this)
+    }
+
+    def close = {
+      session.close
+      dispatch_queue {
+        proxy_sessions.remove(this)
+        consumers.get(session.consumer) match {
+          case Some(proxy) => add_counters(proxy.link, this)
+          case _          => add_counters(Topic.this, this)
+        }
+        producers.get(session.producer.asInstanceOf[BindableDeliveryProducer]) match {
+          case Some(link) => add_counters(link, this)
+          case _          => add_counters(Topic.this, this)
+        }
+      }
+    }
+
+    var enqueue_ts = now
+    def offer(value: Delivery) = {
+      if( session.offer(value) ) {
+        enqueue_ts = now
+        true
+      } else {
+        false
+      }
+    }
+
+    def downstream = session
+    def remaining_capacity = session.remaining_capacity
+    def producer = session.producer
+    def enqueue_size_counter = session.enqueue_size_counter
+    def enqueue_item_counter = session.enqueue_item_counter
+    def consumer = session.consumer
+  }
+
+  case class ProxyDeliveryConsumer(consumer:DeliveryConsumer, link:LinkDTO) extends DeliveryConsumer {
+    def retained() = consumer.retained()
+    def retain() = consumer.retain()
+    def release() = consumer.release()
+    def matches(message: Delivery) = consumer.matches(message)
+    def is_persistent = consumer.is_persistent
+    def dispatch_queue = consumer.dispatch_queue
+    def connect(producer: DeliveryProducer) = {
+      new ProxyDeliverySession(consumer.connect(producer))
+    }
+  }
+
+  val producers = HashMap[BindableDeliveryProducer, LinkDTO]()
+  val consumers = HashMap[DeliveryConsumer, ProxyDeliveryConsumer]()
   var durable_subscriptions = ListBuffer[Queue]()
   var consumer_queues = HashMap[DeliveryConsumer, Queue]()
   var idled_at = 0L
@@ -52,9 +127,87 @@ class Topic(val router:LocalRouter, val 
   override def toString = destination_dto.toString
 
   def virtual_host: VirtualHost = router.virtual_host
+  def now = virtual_host.broker.now
+  def dispatch_queue = virtual_host.dispatch_queue
 
   def slow_consumer_policy = config.slow_consumer_policy.getOrElse("block")
 
+  def status:TopicStatusDTO = {
+    dispatch_queue.assertExecuting()
+
+    val rc = new TopicStatusDTO
+    rc.id = this.id
+    rc.state = "STARTED"
+    rc.state_since = this.created_at
+    rc.config = this.config
+
+    rc.metrics.producer_counter = producer_counter
+    rc.metrics.consumer_counter = consumer_counter
+    rc.metrics.producer_count = producers.size
+    rc.metrics.consumer_counter = consumers.size
+
+    this.durable_subscriptions.foreach { q =>
+      rc.dsubs.add(q.id)
+    }
+
+    def copy(o:LinkDTO) = {
+      val rc = new LinkDTO()
+      rc.id = o.id
+      rc.kind = o.kind
+      rc.label = o.label
+      rc.enqueue_ts = o.enqueue_ts
+      add_counters(rc, o);
+      rc
+    }
+
+    // build the list of producer and consumer links..
+    val producer_links = HashMap[BindableDeliveryProducer, LinkDTO]()
+    val consumers_links = HashMap[DeliveryConsumer, LinkDTO]()
+    this.producers.foreach { case (producer, link) =>
+      val o = copy(link)
+      producer_links.put(producer, o)
+      rc.producers.add(o)
+    }
+    this.consumers.foreach { case (consumer, proxy) =>
+      val o = copy(proxy.link)
+      consumers_links.put(consumer, o)
+      rc.consumers.add(o)
+    }
+
+    // Add in the counters from the live sessions..
+    proxy_sessions.foreach{ session =>
+      val stats = from_session(session)
+      for( link <- producer_links.get(session.producer.asInstanceOf[BindableDeliveryProducer]) ) {
+        add_counters(link, stats)
+      }
+      for( link <- consumers_links.get(session.consumer) ) {
+        add_counters(link, stats)
+      }
+    }
+
+    // Now update the topic counters..
+    rc.metrics.enqueue_item_counter = enqueue_item_counter
+    rc.metrics.enqueue_size_counter = enqueue_size_counter
+    rc.metrics.enqueue_ts = enqueue_ts
+
+    rc.metrics.dequeue_item_counter = dequeue_item_counter
+    rc.metrics.dequeue_size_counter = dequeue_size_counter
+    rc.metrics.dequeue_ts = dequeue_ts
+    consumers_links.values.foreach { link =>
+      rc.metrics.enqueue_item_counter += link.enqueue_item_counter
+      rc.metrics.enqueue_size_counter += link.enqueue_size_counter
+      rc.metrics.enqueue_ts = rc.metrics.enqueue_ts max link.enqueue_ts
+    }
+    producer_links.values.foreach { link =>
+      rc.metrics.dequeue_item_counter += link.enqueue_item_counter
+      rc.metrics.dequeue_size_counter += link.enqueue_size_counter
+      rc.metrics.dequeue_ts = rc.metrics.dequeue_ts max link.enqueue_ts
+    }
+
+    rc
+  }
+
+
   def update(on_completed:Runnable) = {
     refresh_config
     on_completed.run
@@ -80,7 +233,7 @@ class Topic(val router:LocalRouter, val 
         val now = System.currentTimeMillis()
         idled_at = now
         if( auto_delete_after!=0 ) {
-          virtual_host.dispatch_queue.after(auto_delete_after, TimeUnit.SECONDS) {
+          dispatch_queue.after(auto_delete_after, TimeUnit.SECONDS) {
             if( now == idled_at ) {
               router.topic_domain.remove_destination(path, this)
             }
@@ -93,15 +246,11 @@ class Topic(val router:LocalRouter, val 
   }
 
   def bind (destination: DestinationDTO, consumer:DeliveryConsumer) = {
-    destination match {
-      case null=> // unified queue case
-
-        consumers += consumer
-        val list = List(consumer)
-        producers.foreach({ r=>
-          r.bind(list)
-        })
 
+    val target = destination match {
+      case null=>
+        // this is the unified queue case..
+        consumer
       case destination:TopicDestinationDTO=>
         var target = consumer
         slow_consumer_policy match {
@@ -111,64 +260,70 @@ class Topic(val router:LocalRouter, val 
             val queue = router._create_queue(new TempQueueBinding(consumer))
             queue.dispatch_queue.setTargetQueue(consumer.dispatch_queue)
             queue.bind(List(consumer))
-
             consumer_queues += consumer->queue
-            target = queue
+            queue
 
           case "block" =>
             // just have dispatcher dispatch directly to them..
+            consumer
         }
+    }
 
-        consumers += target
-        consumer_counter += 1
-        val list = target :: Nil
-        producers.foreach({ r=>
-          r.bind(list)
-        })
-
+    val link = new LinkDTO()
+    target match {
+      case queue:Queue =>
+        link.kind = "queue"
+        link.id = queue.id
+        link.label = queue.id
+      case _ =>
+        target.connection match {
+          case Some(connection) =>
+            link.kind = "connection"
+            link.id = connection.id.toString
+            link.label = connection.transport.getRemoteAddress.toString
+          case _ =>
+            link.kind = "unknown"
+            link.label = "unknown"
+        }
     }
+
+    val proxy = ProxyDeliveryConsumer(target, link)
+    consumers.put(target, proxy)
+    consumer_counter += 1
+    val list = proxy :: Nil
+    producers.keys.foreach({ r=>
+      r.bind(list)
+    })
     check_idle
   }
 
   def unbind (consumer:DeliveryConsumer, persistent:Boolean) = {
 
-    consumer_queues.remove(consumer) match {
-      case Some(queue)=>
-
-        queue.unbind(List(consumer))
-
-        queue.binding match {
-          case x:TempQueueBinding =>
-
-            val list = List(queue)
-            producers.foreach({ r=>
-              r.unbind(list)
-            })
-            router._destroy_queue(queue.id, null)
-
-        }
-
-      case None=>
-
-        // producers are directly delivering to the consumer..
-        val original = consumers.size
-        consumers -= consumer
-        if( original!= consumers.size ) {
-          val list = List(consumer)
-          producers.foreach({ r=>
-            r.unbind(list)
-          })
-        }
+    for(proxy <- consumers.remove(consumer)) {
+      add_counters(Topic.this, proxy.link)
+      val list = consumer_queues.remove(consumer) match {
+        case Some(queue) =>
+          queue.unbind(List(consumer))
+          queue.binding match {
+            case x:TempQueueBinding =>
+              router._destroy_queue(queue.id, null)
+          }
+          List(queue)
+        case None =>
+          List(consumer)
+      }
+      producers.keys.foreach({ r=>
+        r.unbind(list)
+      })
     }
     check_idle
-
   }
 
   def bind_durable_subscription(destination: DurableSubscriptionDestinationDTO, queue:Queue)  = {
     if( !durable_subscriptions.contains(queue) ) {
       durable_subscriptions += queue
       val list = List(queue)
-      producers.foreach({ r=>
+      producers.keys.foreach({ r=>
         r.bind(list)
       })
       consumer_queues.foreach{case (consumer, q)=>
@@ -184,7 +339,7 @@ class Topic(val router:LocalRouter, val 
     if( durable_subscriptions.contains(queue) ) {
       durable_subscriptions -= queue
       val list = List(queue)
-      producers.foreach({ r=>
+      producers.keys.foreach({ r=>
         r.unbind(list)
       })
       consumer_queues.foreach{case (consumer, q)=>
@@ -197,21 +352,32 @@ class Topic(val router:LocalRouter, val 
   }
 
   def connect (destination:DestinationDTO, producer:BindableDeliveryProducer) = {
-    producers += producer
+    val link = new LinkDTO()
+    producer.connection match {
+      case Some(connection) =>
+        link.kind = "connection"
+        link.id = connection.id.toString
+        link.label = connection.transport.getRemoteAddress.toString
+      case _ =>
+        link.kind = "unknown"
+        link.label = "unknown"
+    }
+    producers.put(producer, link)
     producer_counter += 1
-    producer.bind(consumers.toList ::: durable_subscriptions.toList)
+    producer.bind(consumers.values.toList ::: durable_subscriptions.toList)
     check_idle
   }
 
   def disconnect (producer:BindableDeliveryProducer) = {
-    producers = producers.filterNot( _ == producer )
-    producer.unbind(consumers.toList ::: durable_subscriptions.toList)
+    for(link <- producers.remove(producer) ) {
+      add_counters(this, link)
+    }
     check_idle
   }
 
   def disconnect_producers:Unit ={
-    for( producer <- producers ) {
-      producer.unbind(consumers.toList ::: durable_subscriptions.toList)
+    for( (_, link) <- producers ) {
+      add_counters(this, link)
     }
     producers.clear
     check_idle

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java Fri Aug 26 02:49:45 2011
@@ -31,9 +31,9 @@ import javax.xml.bind.annotation.*;
 public class AggregateQueueMetricsDTO extends QueueMetricsDTO {
 
     /**
-     * The number of queues which where aggregated.
+     * The number of objects which where aggregated.
      */
-    @XmlAttribute(name="queues")
-    public int queues;
+    @XmlAttribute(name="objects")
+    public int objects;
 
 }
\ No newline at end of file

Copied: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateTopicMetricsDTO.java (from r1161956, activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java)
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateTopicMetricsDTO.java?p2=activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateTopicMetricsDTO.java&p1=activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java&r1=1161956&r2=1161969&rev=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateTopicMetricsDTO.java Fri Aug 26 02:49:45 2011
@@ -16,9 +16,10 @@
  */
 package org.apache.activemq.apollo.dto;
 
-import org.codehaus.jackson.annotate.JsonTypeInfo;
-
-import javax.xml.bind.annotation.*;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlRootElement;
 
 /**
  * <p>
@@ -26,14 +27,14 @@ import javax.xml.bind.annotation.*;
  *
  * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
  */
-@XmlRootElement(name="aggregate_queue_metrics")
+@XmlRootElement(name="aggregate_topic_metrics")
 @XmlAccessorType(XmlAccessType.FIELD)
-public class AggregateQueueMetricsDTO extends QueueMetricsDTO {
+public class AggregateTopicMetricsDTO extends TopicMetricsDTO {
 
     /**
-     * The number of queues which where aggregated.
+     * The number of topics which where aggregated.
      */
-    @XmlAttribute(name="queues")
-    public int queues;
+    @XmlAttribute(name="objects")
+    public int objects;
 
 }
\ No newline at end of file

Added: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/DestinationMetricsDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/DestinationMetricsDTO.java?rev=1161969&view=auto
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/DestinationMetricsDTO.java (added)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/DestinationMetricsDTO.java Fri Aug 26 02:49:45 2011
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.apollo.dto;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+
+/**
+ * <p>
+ *     Collects metrics about the status of a destination since the
+ *     time a broker gets started.
+ * </p>
+ *
+ * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
+ */
+@XmlAccessorType(XmlAccessType.FIELD)
+public class DestinationMetricsDTO {
+
+    /**
+     * The current time on the broker machine.  In milliseconds since the epoch.
+     */
+	@XmlAttribute(name="current_time")
+	public long current_time;
+
+    /**
+     * The number of messages that have been sent to the destination.
+     */
+    @XmlAttribute(name="enqueue_item_counter")
+    public long enqueue_item_counter;
+
+    /**
+     * The total size in bytes of messages that have been sent
+     * to the destination
+     */
+    @XmlAttribute(name="enqueue_size_counter")
+    public long enqueue_size_counter;
+
+    /**
+     * The time stamp of when the last message was sent to the destination.
+     */
+    @XmlAttribute(name="enqueue_ts")
+    public long enqueue_ts;
+
+    /**
+     * The number of messages that have been sent to consumers on
+     * the destination.
+     */
+    @XmlAttribute(name="dequeue_item_counter")
+    public long dequeue_item_counter;
+
+    /**
+     * The total size in bytes of messages that have been sent to consumers on
+     * the destination.
+     */
+    @XmlAttribute(name="dequeue_size_counter")
+    public long dequeue_size_counter;
+
+    /**
+     * The time stamp of when the last dequeue to a consumers occurred.
+     */
+    @XmlAttribute(name="dequeue_ts")
+    public long dequeue_ts;
+
+    /**
+     * The total number of producers that have ever sent to
+     * the destination.
+     */
+    @XmlAttribute(name="producer_counter")
+    public long producer_counter;
+
+    /**
+     * The total number of consumers that have ever subscribed to
+     * the queue.
+     */
+    @XmlAttribute(name="consumer_counter")
+    public long consumer_counter;
+
+
+    /**
+     * The current number of producers sending to the destination
+     * the queue.
+     */
+    @XmlAttribute(name="producer_count")
+    public long producer_count;
+
+    /**
+     * The current number of producers consuming from the destination.
+     */
+    @XmlAttribute(name="consumer_count")
+    public long consumer_count;
+
+}

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/LinkDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/LinkDTO.java?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/LinkDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/LinkDTO.java Fri Aug 26 02:49:45 2011
@@ -36,5 +36,23 @@ public class LinkDTO {
     @XmlAttribute
     public String label;
 
+    /**
+     * The number of messages that have been dispatched over the link
+     */
+    @XmlAttribute(name="enqueue_item_counter")
+    public long enqueue_item_counter;
+
+    /**
+     * The total size in bytes of messages that have been dispatched
+     * over the link
+     */
+    @XmlAttribute(name="enqueue_size_counter")
+    public long enqueue_size_counter;
+
+    /**
+     * Timestamp of when a message last went over the link.
+     */
+    @XmlAttribute(name="enqueue_tsr")
+    public long enqueue_ts;
 
 }
\ No newline at end of file

Copied: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerLinkDTO.java (from r1161956, activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerStatusDTO.java)
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerLinkDTO.java?p2=activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerLinkDTO.java&p1=activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerStatusDTO.java&r1=1161956&r2=1161969&rev=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerStatusDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueConsumerLinkDTO.java Fri Aug 26 02:49:45 2011
@@ -21,23 +21,15 @@ import javax.xml.bind.annotation.*;
 /**
  * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
  */
-@XmlRootElement(name="queue_consumer_status")
+@XmlRootElement(name="queue_consumer_link")
 @XmlAccessorType(XmlAccessType.FIELD)
-public class QueueConsumerStatusDTO {
-
-    /**
-     * link to who is consuming from the queue.
-     */
-    public LinkDTO link;
+public class QueueConsumerLinkDTO extends LinkDTO {
 
     public long position = 0;
 
     public int acquired_count;
     public long acquired_size;
 
-    public long total_dispatched_count;
-    public long total_dispatched_size;
-
     public long total_ack_count;
     public long total_nack_count;
 

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueMetricsDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueMetricsDTO.java?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueMetricsDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueMetricsDTO.java Fri Aug 26 02:49:45 2011
@@ -39,56 +39,7 @@ import javax.xml.bind.annotation.XmlRoot
  */
 @XmlRootElement(name="queue_metrics")
 @XmlAccessorType(XmlAccessType.FIELD)
-public class QueueMetricsDTO {
-
-    /**
-     * The current time on the broker machine.  In milliseconds since the epoch.
-     */
-	@XmlAttribute(name="current_time")
-	public long current_time;
-
-    /**
-     * The number of messages that have been added to the queue.
-     * This includes any messages which were recovered from
-     * persistent storage when the broker was first started.
-     */
-    @XmlAttribute(name="enqueue_item_counter")
-    public long enqueue_item_counter;
-
-    /**
-     * The total size in bytes of messages that have been added
-     * to the queue. This includes any messages
-     * which were recovered from persistent storage when the broker
-     * was first started.
-     */
-    @XmlAttribute(name="enqueue_size_counter")
-    public long enqueue_size_counter;
-
-    /**
-     * The time stamp of when the last enqueue occurred.
-     */
-    @XmlAttribute(name="enqueue_ts")
-    public long enqueue_ts;
-
-    /**
-     * The number of messages that have been removed from the queue.
-     */
-    @XmlAttribute(name="dequeue_item_counter")
-    public long dequeue_item_counter;
-
-    /**
-     * The total size in bytes of messages that have been
-     * removed from the queue.
-     */
-    @XmlAttribute(name="dequeue_size_counter")
-    public long dequeue_size_counter;
-
-    /**
-     * The time stamp of when the last dequeue occurred.
-     */
-    @XmlAttribute(name="dequeue_ts")
-    public long dequeue_ts;
-
+public class QueueMetricsDTO extends DestinationMetricsDTO {
     /**
      * The number of messages which expired before they could be processed.
      */
@@ -204,33 +155,4 @@ public class QueueMetricsDTO {
     @XmlAttribute(name="swap_in_size_counter")
     public long swap_in_size_counter;
 
-    /**
-     * The total number of producers that have sent to
-     * the queue.
-     */
-    @XmlAttribute(name="producer_counter")
-    public long producer_counter;
-
-    /**
-     * The total number of consumers that have ever subscribed to
-     * the queue.
-     */
-    @XmlAttribute(name="consumer_counter")
-    public long consumer_counter;
-
-
-    /**
-     * The current number of producers attached to
-     * the queue.
-     */
-    @XmlAttribute(name="producer_count")
-    public long producer_count;
-
-    /**
-     * The current number of consumers attached to
-     * the queue.
-     */
-    @XmlAttribute(name="consumer_count")
-    public long consumer_count;
-
 }

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueStatusDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueStatusDTO.java?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueStatusDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/QueueStatusDTO.java Fri Aug 26 02:49:45 2011
@@ -45,7 +45,6 @@ public class QueueStatusDTO extends Serv
     @XmlElement(name="entry")
     public List<EntryStatusDTO> entries = new ArrayList<EntryStatusDTO>();
 
-
     /**
      * Ids of all connections that are producing to the destination
      */
@@ -56,6 +55,6 @@ public class QueueStatusDTO extends Serv
      * Ids of all connections that are consuming from the destination
      */
     @XmlElement(name="consumer")
-    public List<QueueConsumerStatusDTO> consumers = new ArrayList<QueueConsumerStatusDTO>();
+    public List<QueueConsumerLinkDTO> consumers = new ArrayList<QueueConsumerLinkDTO>();
 
 }
\ No newline at end of file

Copied: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicMetricsDTO.java (from r1161956, activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java)
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicMetricsDTO.java?p2=activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicMetricsDTO.java&p1=activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java&r1=1161956&r2=1161969&rev=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/AggregateQueueMetricsDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicMetricsDTO.java Fri Aug 26 02:49:45 2011
@@ -14,26 +14,24 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.apollo.dto;
 
-import org.codehaus.jackson.annotate.JsonTypeInfo;
+package org.apache.activemq.apollo.dto;
 
-import javax.xml.bind.annotation.*;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlRootElement;
 
 /**
  * <p>
+ *     Collects metrics about the status of a topic since the
+ *     time a broker gets started.
  * </p>
  *
  * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
  */
-@XmlRootElement(name="aggregate_queue_metrics")
+@XmlRootElement(name="topic_metrics")
 @XmlAccessorType(XmlAccessType.FIELD)
-public class AggregateQueueMetricsDTO extends QueueMetricsDTO {
-
-    /**
-     * The number of queues which where aggregated.
-     */
-    @XmlAttribute(name="queues")
-    public int queues;
+public class TopicMetricsDTO extends DestinationMetricsDTO {
 
-}
\ No newline at end of file
+}

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicStatusDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicStatusDTO.java?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicStatusDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/TopicStatusDTO.java Fri Aug 26 02:49:45 2011
@@ -51,17 +51,7 @@ public class TopicStatusDTO extends Serv
     @XmlElement(name="dsub")
     public List<String> dsubs = new ArrayList<String>();
 
-    /**
-     * The total number of producers that have sent to
-     * the topic.
-     */
-    @XmlAttribute(name="producer_counter")
-    public long producer_counter;
+    @XmlElement
+    public TopicMetricsDTO metrics = new TopicMetricsDTO();
 
-    /**
-     * The total number of consumers that have ever subscribed to
-     * the topic.
-     */
-    @XmlAttribute(name="consumer_counter")
-    public long consumer_counter;
 }
\ No newline at end of file

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/VirtualHostStatusDTO.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/VirtualHostStatusDTO.java?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/VirtualHostStatusDTO.java (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/java/org/apache/activemq/apollo/dto/VirtualHostStatusDTO.java Fri Aug 26 02:49:45 2011
@@ -39,36 +39,18 @@ public class VirtualHostStatusDTO extend
     public List<String> topics = new ArrayList<String>();
 
     /**
-     * The number of topics that exist on the virtual host
-     */
-    @XmlAttribute(name="topic_count")
-    public int topic_count;
-
-    /**
      * Ids of all the queues that exist on the broker
      */
     @XmlElement(name="queue")
     public List<String> queues = new ArrayList<String>();
 
     /**
-     * The number of queues that exist on the virtual host
-     */
-    @XmlAttribute(name="queue_count")
-    public int queue_count;
-
-    /**
      * Ids of all the durable subscriptions that exist on the broker
      */
     @XmlElement(name="dsub")
     public List<String> dsubs = new ArrayList<String>();
 
     /**
-     * The number of durable subscriptions that exist on the virtual host
-     */
-    @XmlAttribute(name="dsub_count")
-    public int dsub_count;
-
-    /**
      * Is the virtual host using a store.
      */
     @XmlAttribute(name="store")

Modified: activemq/activemq-apollo/trunk/apollo-dto/src/main/resources/org/apache/activemq/apollo/dto/jaxb.index
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-dto/src/main/resources/org/apache/activemq/apollo/dto/jaxb.index?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-dto/src/main/resources/org/apache/activemq/apollo/dto/jaxb.index (original)
+++ activemq/activemq-apollo/trunk/apollo-dto/src/main/resources/org/apache/activemq/apollo/dto/jaxb.index Fri Aug 26 02:49:45 2011
@@ -42,7 +42,10 @@ NullStoreDTO
 QueueDTO
 TopicDTO
 LinkDTO
-QueueConsumerStatusDTO
+QueueConsumerLinkDTO
 ValueDTO
 StringListDTO
-DataPageDTO
\ No newline at end of file
+DataPageDTO
+AggregateTopicMetricsDTO
+AggregateQueueMetricsDTO
+DestinationMetricsDTO
\ No newline at end of file

Modified: activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala Fri Aug 26 02:49:45 2011
@@ -869,6 +869,8 @@ class OpenwireProtocolHandler extends Pr
 
       val downstream = session_manager.open(producer.dispatch_queue, receive_buffer_size)
       def remaining_capacity = downstream.remaining_capacity
+      def enqueue_item_counter = downstream.accepted_count
+      def enqueue_size_counter = downstream.accepted_size
 
       def close = {
         assert(producer.dispatch_queue.isExecuting)

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=1161969&r1=1161968&r2=1161969&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 Aug 26 02:49:45 2011
@@ -389,6 +389,9 @@ class StompProtocolHandler extends Proto
       val downstream = session_manager.open(producer.dispatch_queue, receive_buffer_size)
 
       def remaining_capacity = downstream.remaining_capacity
+      def enqueue_item_counter = downstream.accepted_count
+      def enqueue_size_counter = downstream.accepted_size
+
 
       def close = {
         assert(producer.dispatch_queue.isExecuting)

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=1161969&r1=1161968&r2=1161969&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 Fri Aug 26 02:49:45 2011
@@ -215,50 +215,110 @@ case class BrokerResource() extends Reso
     rc
   }
 
-  def aggregate_queue_metrics(queue_metrics:Iterable[QueueMetricsDTO]):AggregateQueueMetricsDTO = {
-    queue_metrics.foldLeft(new AggregateQueueMetricsDTO){ (rc, q)=>
-      rc.enqueue_item_counter += q.enqueue_item_counter
-      rc.enqueue_size_counter += q.enqueue_size_counter
-      rc.enqueue_ts = rc.enqueue_ts max q.enqueue_ts
-
-      rc.dequeue_item_counter += q.dequeue_item_counter
-      rc.dequeue_size_counter += q.dequeue_size_counter
-      rc.dequeue_ts = rc.dequeue_ts max q.dequeue_ts
-
-      rc.nack_item_counter += q.nack_item_counter
-      rc.nack_size_counter += q.nack_size_counter
-      rc.nack_ts = rc.nack_ts max q.nack_ts
-
-      rc.expired_item_counter += q.expired_item_counter
-      rc.expired_size_counter += q.expired_size_counter
-      rc.expired_ts = rc.expired_ts max q.expired_ts
-
-      rc.queue_size += q.queue_size
-      rc.queue_items += q.queue_items
-
-      rc.swap_out_item_counter += q.swap_out_item_counter
-      rc.swap_out_size_counter += q.swap_out_size_counter
-      rc.swap_in_item_counter += q.swap_in_item_counter
-      rc.swap_in_size_counter += q.swap_in_size_counter
-
-      rc.swapping_in_size += q.swapping_in_size
-      rc.swapping_out_size += q.swapping_out_size
-
-      rc.swapped_in_items += q.swapped_in_items
-      rc.swapped_in_size += q.swapped_in_size
-
-      rc.swapped_in_size_max += q.swapped_in_size_max
-      rc.producer_counter += q.producer_counter
-      rc.consumer_counter += q.consumer_counter
-      rc.producer_count += q.producer_count
-      rc.consumer_count += q.consumer_count
+  @GET
+  @Path("topic-metrics")
+  def get_topic_metrics(): AggregateTopicMetricsDTO = {
+    val rc:AggregateTopicMetricsDTO = with_broker { broker =>
+      monitoring(broker) {
+        get_topic_metrics(broker)
+      }
+    }
+    rc.current_time = System.currentTimeMillis()
+    rc
+  }
+
+  @GET
+  @Path("dsub-metrics")
+  def get_dsub_metrics(): AggregateQueueMetricsDTO = {
+    val rc:AggregateQueueMetricsDTO = with_broker { broker =>
+      monitoring(broker) {
+        get_dsub_metrics(broker)
+      }
+    }
+    rc.current_time = System.currentTimeMillis()
+    rc
+  }
+
+  def aggregate(queue:AggregateQueueMetricsDTO, topic:AggregateTopicMetricsDTO, dsub:AggregateQueueMetricsDTO):AggregateQueueMetricsDTO = {
+    // zero out the enqueue stats on the dsubs since they will already be accounted for in the topic
+    // stats.
+    dsub.enqueue_item_counter = 0
+    dsub.enqueue_size_counter = 0
+    dsub.enqueue_ts = 0
+    val rc = aggregate_queue_metrics(List(queue, dsub))
+    add_destination_metrics(rc, topic)
+    rc.objects += topic.objects
+    rc.current_time = System.currentTimeMillis()
+    rc
+  }
+
+  @GET
+  @Path("dest-metrics")
+  def get_dest_metrics(): AggregateQueueMetricsDTO = {
+    aggregate(get_queue_metrics(), get_topic_metrics(), get_dsub_metrics())
+  }
+
+  def add_destination_metrics(to:DestinationMetricsDTO, from:DestinationMetricsDTO) = {
+    to.enqueue_item_counter += from.enqueue_item_counter
+    to.enqueue_size_counter += from.enqueue_size_counter
+    to.enqueue_ts = to.enqueue_ts max from.enqueue_ts
+
+    to.dequeue_item_counter += from.dequeue_item_counter
+    to.dequeue_size_counter += from.dequeue_size_counter
+    to.dequeue_ts = to.dequeue_ts max from.dequeue_ts
+
+    to.producer_counter += from.producer_counter
+    to.consumer_counter += from.consumer_counter
+    to.producer_count += from.producer_count
+    to.consumer_count += from.consumer_count
+  }
+
+  def aggregate_queue_metrics(metrics:Iterable[QueueMetricsDTO]):AggregateQueueMetricsDTO = {
+    metrics.foldLeft(new AggregateQueueMetricsDTO){ (memo, metric)=>
+      add_destination_metrics(memo, metric)
+
+      memo.nack_item_counter += metric.nack_item_counter
+      memo.nack_size_counter += metric.nack_size_counter
+      memo.nack_ts = memo.nack_ts max metric.nack_ts
+
+      memo.expired_item_counter += metric.expired_item_counter
+      memo.expired_size_counter += metric.expired_size_counter
+      memo.expired_ts = memo.expired_ts max metric.expired_ts
+
+      memo.queue_size += metric.queue_size
+      memo.queue_items += metric.queue_items
+
+      memo.swap_out_item_counter += metric.swap_out_item_counter
+      memo.swap_out_size_counter += metric.swap_out_size_counter
+      memo.swap_in_item_counter += metric.swap_in_item_counter
+      memo.swap_in_size_counter += metric.swap_in_size_counter
+
+      memo.swapping_in_size += metric.swapping_in_size
+      memo.swapping_out_size += metric.swapping_out_size
 
-      if( q.isInstanceOf[AggregateQueueMetricsDTO] ) {
-        rc.queues += q.asInstanceOf[AggregateQueueMetricsDTO].queues
+      memo.swapped_in_items += metric.swapped_in_items
+      memo.swapped_in_size += metric.swapped_in_size
+
+      memo.swapped_in_size_max += metric.swapped_in_size_max
+
+      if( metric.isInstanceOf[AggregateQueueMetricsDTO] ) {
+        memo.objects += metric.asInstanceOf[AggregateQueueMetricsDTO].objects
       } else {
-        rc.queues += 1
+        memo.objects += 1
       }
-      rc
+      memo
+    }
+  }
+
+  def aggregate_topic_metrics(metrics:Iterable[TopicMetricsDTO]):AggregateTopicMetricsDTO = {
+    metrics.foldLeft(new AggregateTopicMetricsDTO){ (memo, metric)=>
+      add_destination_metrics(memo, metric)
+      if( metric.isInstanceOf[AggregateTopicMetricsDTO] ) {
+        memo.objects += metric.asInstanceOf[AggregateTopicMetricsDTO].objects
+      } else {
+        memo.objects += 1
+      }
+      memo
     }
   }
 
@@ -271,9 +331,40 @@ case class BrokerResource() extends Reso
 
   def get_queue_metrics(host:VirtualHost):FutureResult[AggregateQueueMetricsDTO] = {
     val router:LocalRouter = host
-    val queues: Iterable[Queue] = router.queues_by_id.values
+    val queues: Iterable[Queue] = router.queue_domain.destinations
     val metrics = sync_all(queues) { queue =>
-      get_queue_metrics(queue)
+      queue.get_queue_metrics
+    }
+    metrics.map( x=> Success(aggregate_queue_metrics(x.flatMap(_.success_option))) )
+  }
+
+
+  def get_topic_metrics(broker:Broker):FutureResult[AggregateTopicMetricsDTO] = {
+    val metrics = sync_all(broker.virtual_hosts.values) { host =>
+      get_topic_metrics(host)
+    }
+    metrics.map( x=> Success(aggregate_topic_metrics(x.flatMap(_.success_option)) ))
+  }
+
+  def get_topic_metrics(host:VirtualHost):FutureResult[AggregateTopicMetricsDTO] = {
+    val router:LocalRouter = host
+    val topics: Iterable[Topic] = router.topic_domain.destinations
+    val metrics = topics.map(_.status.metrics)
+    FutureResult(Success(aggregate_topic_metrics(metrics)))
+  }
+
+  def get_dsub_metrics(broker:Broker):FutureResult[AggregateQueueMetricsDTO] = {
+    val metrics = sync_all(broker.virtual_hosts.values) { host =>
+      get_dsub_metrics(host)
+    }
+    metrics.map( x=> Success(aggregate_queue_metrics(x.flatMap(_.success_option)) ))
+  }
+
+  def get_dsub_metrics(host:VirtualHost):FutureResult[AggregateQueueMetricsDTO] = {
+    val router:LocalRouter = host
+    val dsubs: Iterable[Queue] = router.topic_domain.durable_subscriptions_by_id.values
+    val metrics = sync_all(dsubs) { dsub =>
+      dsub.get_queue_metrics
     }
     metrics.map( x=> Success(aggregate_queue_metrics(x.flatMap(_.success_option))) )
   }
@@ -316,15 +407,12 @@ case class BrokerResource() extends Reso
     router.queue_domain.destinations.foreach { node =>
       result.queues.add(node.id)
     }
-    result.queue_count = result.queues.size()
     router.topic_domain.destinations.foreach { node =>
       result.topics.add(node.id)
     }
-    result.topic_count = result.topics.size()
     router.topic_domain.durable_subscriptions_by_id.keys.foreach { id =>
       result.dsubs.add(id)
     }
-    result.dsub_count = result.dsubs.size()
 
     result
   }
@@ -340,6 +428,34 @@ case class BrokerResource() extends Reso
     rc
   }
 
+  @GET @Path("virtual-hosts/{id}/topic-metrics")
+  def virtual_host_topic_metrics(@PathParam("id") id : String): AggregateTopicMetricsDTO = {
+    val rc:AggregateTopicMetricsDTO = with_virtual_host(id) { host =>
+      monitoring(host) {
+        get_topic_metrics(host)
+      }
+    }
+    rc.current_time = System.currentTimeMillis()
+    rc
+  }
+
+  @GET @Path("virtual-hosts/{id}/dsub-metrics")
+  def virtual_host_dsub_metrics(@PathParam("id") id : String): AggregateQueueMetricsDTO = {
+    val rc:AggregateQueueMetricsDTO = with_virtual_host(id) { host =>
+      monitoring(host) {
+        get_dsub_metrics(host)
+      }
+    }
+    rc.current_time = System.currentTimeMillis()
+    rc
+  }
+
+  @GET @Path("virtual-hosts/{id}/dest-metrics")
+  def virtual_host_dest_metrics(@PathParam("id") id : String): AggregateQueueMetricsDTO = {
+    aggregate(virtual_host_queue_metrics(id), virtual_host_topic_metrics(id), virtual_host_dsub_metrics(id))
+  }
+
+
   @GET @Path("virtual-hosts/{id}/store")
   def store(@PathParam("id") id : String):StoreStatusDTO = {
     with_virtual_host(id) { host =>
@@ -357,22 +473,6 @@ case class BrokerResource() extends Reso
     }
   }
 
-  def link(connection:BrokerConnection) = {
-    val link = new LinkDTO()
-    link.kind = "connection"
-    link.id = connection.id.toString
-    link.label = connection.transport.getRemoteAddress.toString
-    link
-  }
-
-  def link(queue:Queue) = {
-    val link = new LinkDTO()
-    link.kind = "queue"
-    link.id = queue.id
-    link.label = queue.id
-    link
-  }
-
   class JosqlHelper {
 
     def get(o:AnyRef, name:String):AnyRef = {
@@ -564,101 +664,14 @@ case class BrokerResource() extends Reso
     }
   }
 
-  def status(node: Topic): FutureResult[TopicStatusDTO] = {
-    monitoring(node) {
-      val rc = new TopicStatusDTO
-      rc.id = node.id
-      rc.state = "STARTED"
-      rc.state_since = node.created_at
-      rc.config = node.config
-      rc.producer_counter = node.producer_counter
-      rc.consumer_counter = node.consumer_counter
-
-      node.durable_subscriptions.foreach {
-        q =>
-          rc.dsubs.add(q.id)
-      }
-      node.consumers.foreach {
-        consumer =>
-          consumer match {
-            case queue: Queue =>
-              rc.consumers.add(link(queue))
-            case _ =>
-              consumer.connection.foreach {
-                c =>
-                  rc.consumers.add(link(c))
-              }
-          }
-      }
-      node.producers.flatMap(_.connection).foreach {
-        connection =>
-          rc.producers.add(link(connection))
-      }
-
-      rc
-    }
+  def status(node: Topic): FutureResult[TopicStatusDTO] = monitoring(node) {
+    node.status
   }
 
   def status(q:Queue, entries:Boolean=false) = monitoring(q) {
-    val rc = new QueueStatusDTO
-    rc.id = q.id
-    rc.state = q.service_state.toString
-    rc.state_since = q.service_state.since
-    rc.binding = q.binding.binding_dto
-    rc.config = q.config
-    rc.metrics = get_queue_metrics(q)
-    rc.metrics.current_time = System.currentTimeMillis()
-
-    if( entries ) {
-      var cur = q.head_entry
-      while( cur!=null ) {
-
-        val e = new EntryStatusDTO
-        e.seq = cur.seq
-        e.count = cur.count
-        e.size = cur.size
-        e.consumer_count = cur.parked.size
-        e.is_prefetched = cur.is_prefetched
-        e.state = cur.label
-
-        rc.entries.add(e)
-
-        cur = if( cur == q.tail_entry ) {
-          null
-        } else {
-          cur.nextOrTail
-        }
-      }
-    }
-
-    q.inbound_sessions.flatMap( _.producer.connection ).foreach { connection=>
-      rc.producers.add(link(connection))
-    }
-    q.all_subscriptions.valuesIterator.toSeq.foreach{ sub =>
-      val status = new QueueConsumerStatusDTO
-      sub.consumer.connection.foreach(x=> status.link = link(x))
-      status.position = sub.pos.seq
-      status.total_dispatched_count = sub.total_dispatched_count
-      status.total_dispatched_size = sub.total_dispatched_size
-      status.total_ack_count = sub.total_ack_count
-      status.total_nack_count = sub.total_nack_count
-      status.acquired_size = sub.acquired_size
-      status.acquired_count = sub.acquired_count
-      status.waiting_on = if( sub.full ) {
-        "ack"
-      } else if( sub.pos.is_tail ) {
-        "producer"
-      } else if( !sub.pos.is_loaded ) {
-        "load"
-      } else {
-        "dispatch"
-      }
-      rc.consumers.add(status)
-    }
-    rc
+    q.status(entries)
   }
 
-
   @GET @Path("connectors")
   @Produces(Array("application/json"))
   def connectors(@QueryParam("f") f:java.util.List[String], @QueryParam("q") q:String,
@@ -778,50 +791,4 @@ case class BrokerResource() extends Reso
     }
   }
 
-  def get_queue_metrics(q:Queue):QueueMetricsDTO = {
-    val rc = new QueueMetricsDTO
-
-    rc.enqueue_item_counter = q.enqueue_item_counter
-    rc.enqueue_size_counter = q.enqueue_size_counter
-    rc.enqueue_ts = q.enqueue_ts
-
-    rc.dequeue_item_counter = q.dequeue_item_counter
-    rc.dequeue_size_counter = q.dequeue_size_counter
-    rc.dequeue_ts = q.dequeue_ts
-
-    rc.nack_item_counter = q.nack_item_counter
-    rc.nack_size_counter = q.nack_size_counter
-    rc.nack_ts = q.nack_ts
-
-    rc.expired_item_counter = q.expired_item_counter
-    rc.expired_size_counter = q.expired_size_counter
-    rc.expired_ts = q.expired_ts
-
-    rc.queue_size = q.queue_size
-    rc.queue_items = q.queue_items
-
-    rc.swap_out_item_counter = q.swap_out_item_counter
-    rc.swap_out_size_counter = q.swap_out_size_counter
-    rc.swap_in_item_counter = q.swap_in_item_counter
-    rc.swap_in_size_counter = q.swap_in_size_counter
-
-    rc.swapping_in_size = q.swapping_in_size
-    rc.swapping_out_size = q.swapping_out_size
-
-    rc.swapped_in_items = q.swapped_in_items
-    rc.swapped_in_size = q.swapped_in_size
-
-    rc.swapped_in_size_max = q.swapped_in_size_max
-
-    rc.producer_counter = q.producer_counter
-    rc.consumer_counter = q.consumer_counter
-
-    rc.producer_count = q.producers.size
-    rc.consumer_count = q.all_subscriptions.size
-
-    rc
-  }
-
-
-
 }

Modified: activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/QueueStatusDTO.jade
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/QueueStatusDTO.jade?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/QueueStatusDTO.jade (original)
+++ activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/QueueStatusDTO.jade Fri Aug 26 02:49:45 2011
@@ -74,28 +74,33 @@ h3 Producers
 p total producers ever : #{metrics.producer_counter}
 ul
   - for( x <- producers )
-    - x.kind match
-      - case "connection" =>
-        li.producer
+    li.producer
+      - x.kind match
+        - case "queue" =>
+          a(href={ path("../../queues/"+x.id+".html") }) #{x.label}
+        - case "connection" =>
           a(href={ path("../../../../connections/"+x.id+".html") }) #{x.label}
-      - case _ =>
+        - case _ =>
+      p dispatched: #{x.enqueue_item_counter} messages (#{memory(x.enqueue_size_counter)}), #{uptime(x.enqueue_ts)} ago
 
 
 h3 Consumers
 p total consumers ever : #{metrics.consumer_counter}
 ul
-  - for( consumer <- consumers )
-    - import consumer._
+  - for( x <- consumers )
     li.consumer
-      - if( link !=null )
-        a(href={ path("../../../../connections/"+link.id+".html") }) #{link.label}
-
-      p next message seq: #{position}
-      p acquired: #{acquired_count} messages (#{memory(acquired_size)})
-      p dispatched: #{total_dispatched_count} messages (#{memory(total_dispatched_size)})
-      p acks: #{total_ack_count} messages
-      p naks: #{total_nack_count} messages
-      p waiting on: #{waiting_on}
+      - x.kind match
+        - case "queue" =>
+          a(href={ path("../../queues/"+x.id+".html") }) #{x.label}
+        - case "connection" =>
+          a(href={ path("../../../../connections/"+x.id+".html") }) #{x.label}
+        - case _ =>
+      p dispatched: #{x.enqueue_item_counter} messages (#{memory(x.enqueue_size_counter)}), #{uptime(x.enqueue_ts)} ago
+      p next message seq: #{x.position}
+      p acquired: #{x.acquired_count} messages (#{memory(x.acquired_size)})
+      p acks: #{x.total_ack_count} messages
+      p naks: #{x.total_nack_count} messages
+      p waiting on: #{x.waiting_on}
 
 - if ( entries.isEmpty )
   h2

Modified: activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/TopicStatusDTO.jade
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/TopicStatusDTO.jade?rev=1161969&r1=1161968&r2=1161969&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/TopicStatusDTO.jade (original)
+++ activemq/activemq-apollo/trunk/apollo-web/src/main/webapp/WEB-INF/org/apache/activemq/apollo/dto/TopicStatusDTO.jade Fri Aug 26 02:49:45 2011
@@ -26,27 +26,30 @@ p state: #{state} #{ uptime(state_since)
 h3 Topic Domain
 
 h4 Producers
-p total producers ever : #{producer_counter}
+p total producers ever : #{metrics.producer_counter}
 ul
   - for( x <- producers )
-    - x.kind match
-      - case "connection" =>
-        li
+    li.producer
+      - x.kind match
+        - case "queue" =>
+          a(href={ path("../../queues/"+x.id+".html") }) #{x.label}
+        - case "connection" =>
           a(href={ path("../../../../connections/"+x.id+".html") }) #{x.label}
-      - case _ =>
+        - case _ =>
+      p dispatched: #{x.enqueue_item_counter} messages (#{memory(x.enqueue_size_counter)}), #{uptime(x.enqueue_ts)} ago
 
 h4 Consumers
-p total consumers ever : #{consumer_counter}
+p total consumers ever : #{metrics.producer_counter}
 ul
   - for( x <- consumers )
-    - x.kind match
-      - case "queue" =>
-        li
+    li.consumer
+      - x.kind match
+        - case "queue" =>
           a(href={ path("../../queues/"+x.id+".html") }) #{x.label}
-      - case "connection" =>
-        li
+        - case "connection" =>
           a(href={ path("../../../../connections/"+x.id+".html") }) #{x.label}
-      - case _ =>
+        - case _ =>
+      p dispatched: #{x.enqueue_item_counter} messages (#{memory(x.enqueue_size_counter)}), #{uptime(x.enqueue_ts)} ago
 
 h4 Durable Subscribers
 ul



Mime
View raw message