Custom mailbox that is bounded and instrumented.

[controller.git] / opendaylight / md-sal / sal-clustering-commons / src / main / java / org / opendaylight / controller / common / actor / MeteredBoundedMailbox.java

diff --git a/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/common/actor/MeteredBoundedMailbox.java b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/common/actor/MeteredBoundedMailbox.java
new file mode 100644 (file)
index 0000000..6464315
--- /dev/null
+++ b/opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/common/actor/MeteredBoundedMailbox.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.common.actor;
+
+import akka.actor.ActorPath;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.dispatch.BoundedMailbox;
+import akka.dispatch.MailboxType;
+import akka.dispatch.MessageQueue;
+import akka.dispatch.ProducesMessageQueue;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.MetricRegistry;
+import com.google.common.base.Preconditions;
+import com.typesafe.config.Config;
+import org.opendaylight.controller.common.reporting.MetricsReporter;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.util.concurrent.TimeUnit;
+
+public class MeteredBoundedMailbox implements MailboxType, ProducesMessageQueue<BoundedMailbox.MessageQueue> {
+
+    private MeteredMessageQueue queue;
+    private Integer capacity;
+    private FiniteDuration pushTimeOut;
+    private ActorPath actorPath;
+    private MetricsReporter reporter;
+
+    private final String QUEUE_SIZE = "queue-size";
+    private final Long DEFAULT_TIMEOUT = 10L;
+
+    public MeteredBoundedMailbox(ActorSystem.Settings settings, Config config) {
+        Preconditions.checkArgument( config.hasPath("mailbox-capacity"), "Missing configuration [mailbox-capacity]" );
+        this.capacity = config.getInt("mailbox-capacity");
+        Preconditions.checkArgument( this.capacity > 0, "mailbox-capacity must be > 0");
+
+        Long timeout = -1L;
+        if ( config.hasPath("mailbox-push-timeout-time") ){
+            timeout = config.getDuration("mailbox-push-timeout-time", TimeUnit.NANOSECONDS);
+        } else {
+            timeout = DEFAULT_TIMEOUT;
+        }
+        Preconditions.checkArgument( timeout > 0, "mailbox-push-timeout-time must be > 0");
+        this.pushTimeOut = new FiniteDuration(timeout, TimeUnit.NANOSECONDS);
+
+        reporter = MetricsReporter.getInstance();
+    }
+
+
+    @Override
+    public MessageQueue create(final scala.Option<ActorRef> owner, scala.Option<ActorSystem> system) {
+        this.queue = new MeteredMessageQueue(this.capacity, this.pushTimeOut);
+        monitorQueueSize(owner, this.queue);
+        return this.queue;
+    }
+
+    private void monitorQueueSize(scala.Option<ActorRef> owner, final MeteredMessageQueue monitoredQueue) {
+        if (owner.isEmpty()) {
+            return; //there's no actor to monitor
+        }
+        actorPath = owner.get().path();
+        MetricRegistry registry = reporter.getMetricsRegistry();
+
+        String actorName = registry.name(actorPath.toString(), QUEUE_SIZE);
+
+        if (registry.getMetrics().containsKey(actorName))
+            return; //already registered
+
+        reporter.getMetricsRegistry().register(actorName,
+                new Gauge<Integer>() {
+                    @Override
+                    public Integer getValue() {
+                        return monitoredQueue.size();
+                    }
+                });
+    }
+
+
+    public static class MeteredMessageQueue extends BoundedMailbox.MessageQueue {
+
+        public MeteredMessageQueue(int capacity, FiniteDuration pushTimeOut) {
+            super(capacity, pushTimeOut);
+        }
+    }
+
+}
+