You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/06/10 19:38:19 UTC
svn commit: r1348629 - in
/qpid/branches/java-config-and-management/qpid/java: broker-plugins/jmx/
broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/
broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/
broker/src/main/java/org...
Author: robbie
Date: Sun Jun 10 17:38:18 2012
New Revision: 1348629
URL: http://svn.apache.org/viewvc?rev=1348629&view=rev
Log:
NO-JIRA: move processing of whether notifications are due into the queue itself, add listener to instruct the Queue MBean to emit any required JMX notifications, remove old Queue/VirtualHost and previously missed abstract Connection MBean code and associated tests
Added:
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/QueueNotificationListener.java
Removed:
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AMQBrokerManagerMBean.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AMQQueueMBean.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AbstractAMQManagedConnectionObject.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/management/VirtualHostMBean.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueNotificationListener.java
qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java
qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
Modified:
qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/MANIFEST.MF
qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/AMQManagedObject.java
qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostMBean.java
qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Queue.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/impl/QueueImpl.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/OsgiSystemPackages.properties
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/NotificationCheck.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
qpid/branches/java-config-and-management/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagedQueueMBeanTest.java
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/MANIFEST.MF
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/MANIFEST.MF?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/MANIFEST.MF (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/MANIFEST.MF Sun Jun 10 17:38:18 2012
@@ -57,6 +57,7 @@ Import-Package: org.apache.qpid,
javax.servlet,
javax.servlet.http,
javax.management;version=1.0.0,
+ javax.management.monitor;version=1.0.0,
javax.management.openmbean;version=1.0.0,
org.osgi.util.tracker;version=1.0.0,
org.osgi.framework;version=1.3
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/AMQManagedObject.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/AMQManagedObject.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/AMQManagedObject.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/AMQManagedObject.java Sun Jun 10 17:38:18 2012
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.server.jmx;
+import java.util.concurrent.atomic.AtomicLong;
+
import javax.management.ListenerNotFoundException;
import javax.management.NotCompliantMBeanException;
import javax.management.NotificationBroadcaster;
@@ -38,7 +40,7 @@ public abstract class AMQManagedObject e
{
private NotificationBroadcasterSupport _broadcaster = new NotificationBroadcasterSupport();
- private long _notificationSequenceNumber = 0;
+ private AtomicLong _notificationSequenceNumber = new AtomicLong();
protected AMQManagedObject(Class<?> managementInterface, String typeName, ManagedObjectRegistry registry)
throws NotCompliantMBeanException
@@ -73,22 +75,9 @@ public abstract class AMQManagedObject e
return _broadcaster;
}
- /**
- * sequence number for notifications
- */
- protected long getNotificationSequenceNumber()
- {
- return _notificationSequenceNumber;
- }
-
- protected void setNotificationSequenceNumber(long notificationSequenceNumber)
- {
- _notificationSequenceNumber = notificationSequenceNumber;
- }
-
protected long incrementAndGetSequenceNumber()
{
- return ++_notificationSequenceNumber;
+ return _notificationSequenceNumber.incrementAndGet();
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java Sun Jun 10 17:38:18 2012
@@ -27,8 +27,11 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.JMException;
+import javax.management.MBeanNotificationInfo;
+import javax.management.Notification;
import javax.management.ObjectName;
import javax.management.OperationsException;
+import javax.management.monitor.MonitorNotification;
import javax.management.openmbean.ArrayType;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.CompositeDataSupport;
@@ -47,11 +50,13 @@ import org.apache.qpid.server.message.AM
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.QueueNotificationListener;
import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.queue.NotificationCheck;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.queue.QueueEntryVisitor;
-public class QueueMBean extends AMQManagedObject implements ManagedQueue
+public class QueueMBean extends AMQManagedObject implements ManagedQueue, QueueNotificationListener
{
private static final String[] VIEW_MSGS_COMPOSITE_ITEM_NAMES_DESC_ARRAY =
VIEW_MSGS_COMPOSITE_ITEM_NAMES_DESC.toArray(new String[VIEW_MSGS_COMPOSITE_ITEM_NAMES_DESC.size()]);
@@ -118,6 +123,7 @@ public class QueueMBean extends AMQManag
_queue = queue;
_vhostMBean = virtualHostMBean;
register();
+ _queue.setNotificationListener(this);
}
public ManagedObject getParentObject()
@@ -630,6 +636,31 @@ public class QueueMBean extends AMQManag
return _entry;
}
}
+
+ @Override
+ public void notifyClients(NotificationCheck notification, Queue queue, String notificationMsg)
+ {
+ notificationMsg = notification.name() + " " + notificationMsg;
+
+ Notification note = new Notification(MonitorNotification.THRESHOLD_VALUE_EXCEEDED, this,
+ incrementAndGetSequenceNumber(), System.currentTimeMillis(), notificationMsg);
+
+ getBroadcaster().sendNotification(note);
+ }
+
+ /**
+ * returns Notifications sent by this MBean.
+ */
+ @Override
+ public MBeanNotificationInfo[] getNotificationInfo()
+ {
+ String[] notificationTypes = new String[] { MonitorNotification.THRESHOLD_VALUE_EXCEEDED };
+ String name = MonitorNotification.class.getName();
+ String description = "Either Message count or Queue depth or Message size has reached threshold high value";
+ MBeanNotificationInfo info1 = new MBeanNotificationInfo(notificationTypes, name, description);
+
+ return new MBeanNotificationInfo[] { info1 };
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostMBean.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostMBean.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostMBean.java Sun Jun 10 17:38:18 2012
@@ -52,13 +52,13 @@ public class VirtualHostMBean extends AM
{
super(ManagedVirtualHost.class, ManagedVirtualHost.TYPE, registry);
_virtualHost = virtualHost;
- //register();
virtualHost.addChangeListener(this);
initQueues();
initExchanges();
initConnections();
+ //This is the actual JMX bean for this 'VirtualHostMBean', leave it alone.
_managerMBean = new VirtualHostManagerMBean(this);
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/VirtualHostManagerMBean.java Sun Jun 10 17:38:18 2012
@@ -33,7 +33,6 @@ import javax.management.MBeanException;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
-import org.apache.qpid.AMQException;
import org.apache.qpid.management.common.mbeans.ManagedBroker;
import org.apache.qpid.management.common.mbeans.ManagedQueue;
import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Queue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Queue.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Queue.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Queue.java Sun Jun 10 17:38:18 2012
@@ -25,7 +25,6 @@ import java.util.Collection;
import java.util.Collections;
import org.apache.qpid.server.queue.QueueEntryVisitor;
-
public interface Queue extends ConfiguredObject
{
public static final String BINDING_COUNT = "bindingCount";
@@ -143,4 +142,6 @@ public interface Queue extends Configure
void visit(QueueEntryVisitor visitor);
void delete();
+
+ void setNotificationListener(QueueNotificationListener listener);
}
Added: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/QueueNotificationListener.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/QueueNotificationListener.java?rev=1348629&view=auto
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/QueueNotificationListener.java (added)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/QueueNotificationListener.java Sun Jun 10 17:38:18 2012
@@ -0,0 +1,28 @@
+/*
+ *
+ * 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.qpid.server.model;
+
+import org.apache.qpid.server.queue.NotificationCheck;
+
+public interface QueueNotificationListener
+{
+ void notifyClients(NotificationCheck notification, Queue queue, String notificationMsg);
+}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java Sun Jun 10 17:38:18 2012
@@ -34,12 +34,13 @@ import org.apache.qpid.server.model.Cons
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.QueueNotificationListener;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.Statistics;
import org.apache.qpid.server.queue.*;
import org.apache.qpid.server.subscription.Subscription;
-final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.SubscriptionRegistrationListener
+final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.SubscriptionRegistrationListener, AMQQueue.NotificationListener
{
static final Map<String, String> ATTRIBUTE_MAPPINGS = new HashMap<String, String>();
@@ -69,6 +70,7 @@ final class QueueAdapter extends Abstrac
private final VirtualHostAdapter _vhost;
private QueueStatisticsAdapter _statistics;
+ private QueueNotificationListener _queueNotificationListener;
public QueueAdapter(final VirtualHostAdapter virtualHostAdapter, final AMQQueue queue)
{
@@ -80,6 +82,7 @@ final class QueueAdapter extends Abstrac
_queue.addSubscriptionRegistrationListener(this);
populateConsumers();
_statistics = new QueueStatisticsAdapter(queue);
+ _queue.setNotificationListener(this);
}
private void populateConsumers()
@@ -645,4 +648,19 @@ final class QueueAdapter extends Abstrac
}
}
+ @Override
+ public void setNotificationListener(QueueNotificationListener listener)
+ {
+ _queueNotificationListener = listener;
+ }
+
+ @Override
+ public void notifyClients(NotificationCheck notification, AMQQueue queue, String notificationMsg)
+ {
+ QueueNotificationListener listener = _queueNotificationListener;
+ if(listener != null)
+ {
+ listener.notifyClients(notification, this, notificationMsg);
+ }
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/impl/QueueImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/impl/QueueImpl.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/impl/QueueImpl.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/model/impl/QueueImpl.java Sun Jun 10 17:38:18 2012
@@ -28,6 +28,7 @@ import org.apache.qpid.server.model.Bind
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.QueueNotificationListener;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.Statistics;
import org.apache.qpid.server.model.Consumer;
@@ -97,4 +98,10 @@ class QueueImpl extends AbstractConfigur
{
throw new UnsupportedOperationException(); // TODO
}
+
+ @Override
+ public void setNotificationListener(QueueNotificationListener listener)
+ {
+ // TODO - implement
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/OsgiSystemPackages.properties
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/OsgiSystemPackages.properties?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/OsgiSystemPackages.properties (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/OsgiSystemPackages.properties Sun Jun 10 17:38:18 2012
@@ -33,6 +33,7 @@ javax.management.openmbean=1.0.0
javax.management=1.0.0
javax.management.remote.rmi=1.0.0
javax.management.remote=1.0.0
+javax.management.monitor=1.0.0
javax.crypto=1
javax.crypto.spec=1
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java Sun Jun 10 17:38:18 2012
@@ -28,8 +28,6 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeReferrer;
import org.apache.qpid.server.logging.LogSubject;
-import org.apache.qpid.server.management.Managable;
-import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.AuthorizationHolder;
import org.apache.qpid.server.store.TransactionLogResource;
@@ -41,9 +39,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-public interface AMQQueue extends Managable, Comparable<AMQQueue>, ExchangeReferrer, TransactionLogResource, BaseQueue,
+public interface AMQQueue extends Comparable<AMQQueue>, ExchangeReferrer, TransactionLogResource, BaseQueue,
QueueConfig
{
+ public interface NotificationListener
+ {
+ void notifyClients(NotificationCheck notification, AMQQueue queue, String notificationMsg);
+ }
+
boolean getDeleteOnNoConsumers();
void setDeleteOnNoConsumers(boolean b);
@@ -287,8 +290,6 @@ public interface AMQQueue extends Managa
ConfigurationPlugin getConfiguration();
- ManagedObject getManagedObject();
-
void setExclusive(boolean exclusive) throws AMQException;
/**
@@ -308,4 +309,5 @@ public interface AMQQueue extends Managa
*/
public void setMaximumDeliveryCount(final int maximumDeliveryCount);
+ void setNotificationListener(NotificationListener listener);
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java Sun Jun 10 17:38:18 2012
@@ -20,10 +20,7 @@
*/
package org.apache.qpid.server.queue;
-import org.apache.log4j.Logger;
-import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.server.exchange.DefaultExchangeRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.ArrayList;
@@ -34,8 +31,6 @@ import java.util.concurrent.ConcurrentMa
public class DefaultQueueRegistry implements QueueRegistry
{
- private static final Logger LOGGER = Logger.getLogger(DefaultExchangeRegistry.class);
-
private ConcurrentMap<AMQShortString, AMQQueue> _queueMap = new ConcurrentHashMap<AMQShortString, AMQQueue>();
private final VirtualHost _virtualHost;
@@ -113,16 +108,16 @@ public class DefaultQueueRegistry implem
for (final AMQQueue queue : getQueues())
{
queue.stop();
- try
- {
- //TODO: this needs updated to use the
- //listener model, the managed object being
- //accessed here is no longer used
- queue.getManagedObject().unregister();
- }
- catch (AMQException e)
+
+ //TODO: this is a bit of a hack, what if the listeners aren't aware
+ //that we are just unregistering the MBean because of HA, and aren't
+ //actually removing the queue as such.
+ synchronized (_listeners)
{
- LOGGER.warn("Failed to unregister mbean", e);
+ for(RegistryChangeListener listener : _listeners)
+ {
+ listener.queueUnregistered(queue);
+ }
}
}
_queueMap.clear();
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/NotificationCheck.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/NotificationCheck.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/NotificationCheck.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/NotificationCheck.java Sun Jun 10 17:38:18 2012
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.queue;
+import org.apache.log4j.Logger;
import org.apache.qpid.server.message.ServerMessage;
public enum NotificationCheck
@@ -27,13 +28,16 @@ public enum NotificationCheck
MESSAGE_COUNT_ALERT
{
- public boolean notifyIfNecessary(ServerMessage msg, AMQQueue queue, QueueNotificationListener listener)
+ public boolean notifyIfNecessary(ServerMessage<?> msg, AMQQueue queue, AMQQueue.NotificationListener listener)
{
int msgCount;
final long maximumMessageCount = queue.getMaximumMessageCount();
if (maximumMessageCount!= 0 && (msgCount = queue.getMessageCount()) >= maximumMessageCount)
{
- listener.notifyClients(this, queue, msgCount + ": Maximum count on queue threshold ("+ maximumMessageCount +") breached.");
+ String notificationMsg = msgCount + ": Maximum count on queue threshold ("+ maximumMessageCount +") breached.";
+
+ logNotification(this, queue, notificationMsg);
+ listener.notifyClients(this, queue, notificationMsg);
return true;
}
return false;
@@ -41,7 +45,7 @@ public enum NotificationCheck
},
MESSAGE_SIZE_ALERT(true)
{
- public boolean notifyIfNecessary(ServerMessage msg, AMQQueue queue, QueueNotificationListener listener)
+ public boolean notifyIfNecessary(ServerMessage<?> msg, AMQQueue queue, AMQQueue.NotificationListener listener)
{
final long maximumMessageSize = queue.getMaximumMessageSize();
if(maximumMessageSize != 0)
@@ -50,10 +54,12 @@ public enum NotificationCheck
long messageSize;
messageSize = (msg == null) ? 0 : msg.getSize();
-
if (messageSize >= maximumMessageSize)
{
- listener.notifyClients(this, queue, messageSize + "b : Maximum message size threshold ("+ maximumMessageSize +") breached. [Message ID=" + msg.getMessageNumber() + "]");
+ String notificationMsg = messageSize + "b : Maximum message size threshold ("+ maximumMessageSize +") breached. [Message ID=" + msg.getMessageNumber() + "]";
+
+ logNotification(this, queue, notificationMsg);
+ listener.notifyClients(this, queue, notificationMsg);
return true;
}
}
@@ -63,7 +69,7 @@ public enum NotificationCheck
},
QUEUE_DEPTH_ALERT
{
- public boolean notifyIfNecessary(ServerMessage msg, AMQQueue queue, QueueNotificationListener listener)
+ public boolean notifyIfNecessary(ServerMessage<?> msg, AMQQueue queue, AMQQueue.NotificationListener listener)
{
// Check for threshold queue depth in bytes
final long maximumQueueDepth = queue.getMaximumQueueDepth();
@@ -74,7 +80,10 @@ public enum NotificationCheck
if (queueDepth >= maximumQueueDepth)
{
- listener.notifyClients(this, queue, (queueDepth>>10) + "Kb : Maximum queue depth threshold ("+(maximumQueueDepth>>10)+"Kb) breached.");
+ String notificationMsg = (queueDepth>>10) + "Kb : Maximum queue depth threshold ("+(maximumQueueDepth>>10)+"Kb) breached.";
+
+ logNotification(this, queue, notificationMsg);
+ listener.notifyClients(this, queue, notificationMsg);
return true;
}
}
@@ -84,7 +93,7 @@ public enum NotificationCheck
},
MESSAGE_AGE_ALERT
{
- public boolean notifyIfNecessary(ServerMessage msg, AMQQueue queue, QueueNotificationListener listener)
+ public boolean notifyIfNecessary(ServerMessage<?> msg, AMQQueue queue, AMQQueue.NotificationListener listener)
{
final long maxMessageAge = queue.getMaximumMessageAge();
@@ -97,7 +106,10 @@ public enum NotificationCheck
if(firstArrivalTime < thresholdTime)
{
long oldestAge = currentTime - firstArrivalTime;
- listener.notifyClients(this, queue, (oldestAge/1000) + "s : Maximum age on queue threshold ("+(maxMessageAge /1000)+"s) breached.");
+ String notificationMsg = (oldestAge/1000) + "s : Maximum age on queue threshold ("+(maxMessageAge /1000)+"s) breached.";
+
+ logNotification(this, queue, notificationMsg);
+ listener.notifyClients(this, queue, notificationMsg);
return true;
}
@@ -109,6 +121,8 @@ public enum NotificationCheck
}
;
+ private static final Logger LOGGER = Logger.getLogger(NotificationCheck.class);
+
private final boolean _messageSpecific;
NotificationCheck()
@@ -126,6 +140,11 @@ public enum NotificationCheck
return _messageSpecific;
}
- public abstract boolean notifyIfNecessary(ServerMessage msg, AMQQueue queue, QueueNotificationListener listener);
+ public abstract boolean notifyIfNecessary(ServerMessage<?> msg, AMQQueue queue, AMQQueue.NotificationListener listener);
+ //A bit of a hack, only for use until we do the logging listener
+ private static void logNotification(NotificationCheck notification, AMQQueue queue, String notificationMsg)
+ {
+ LOGGER.info(notification.name() + " On Queue " + queue.getNameShortString() + " - " + notificationMsg);
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Sun Jun 10 17:38:18 2012
@@ -33,7 +33,7 @@ import java.util.concurrent.atomic.Atomi
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-import javax.management.JMException;
+import javax.management.Notification;
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
@@ -53,8 +53,6 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.actors.QueueActor;
import org.apache.qpid.server.logging.messages.QueueMessages;
import org.apache.qpid.server.logging.subjects.QueueLogSubject;
-import org.apache.qpid.server.management.AMQQueueMBean;
-import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -177,7 +175,6 @@ public class SimpleAMQQueue implements A
private LogSubject _logSubject;
private LogActor _logActor;
- private AMQQueueMBean _managedObject;
private static final String SUB_FLUSH_RUNNER = "SUB_FLUSH_RUNNER";
private boolean _nolocal;
@@ -198,6 +195,8 @@ public class SimpleAMQQueue implements A
private final Collection<SubscriptionRegistrationListener> _subscriptionListeners =
new ArrayList<SubscriptionRegistrationListener>();
+ private AMQQueue.NotificationListener _notificationListener;
+ private final long[] _lastNotificationTimes = new long[NotificationCheck.values().length];
protected SimpleAMQQueue(UUID id, AMQShortString name, boolean durable, AMQShortString owner, boolean autoDelete, boolean exclusive, VirtualHost virtualHost, Map<String,Object> arguments)
{
@@ -263,16 +262,6 @@ public class SimpleAMQQueue implements A
getConfigStore().addConfiguredObject(this);
- try
- {
- _managedObject = new AMQQueueMBean(this);
- _managedObject.register();
- }
- catch (JMException e)
- {
- _logger.error("AMQQueue MBean creation has failed ", e);
- }
-
if(arguments != null && arguments.containsKey(QPID_GROUP_HEADER_KEY))
{
if(arguments.containsKey(QPID_SHARED_MSG_GROUP) && String.valueOf(arguments.get(QPID_SHARED_MSG_GROUP)).equals("1"))
@@ -771,10 +760,7 @@ public class SimpleAMQQueue implements A
}
}
- if(_managedObject != null)
- {
- _managedObject.checkForNotification(entry.getMessage());
- }
+ checkForNotification(entry.getMessage());
if(action != null)
{
@@ -1683,12 +1669,6 @@ public class SimpleAMQQueue implements A
txn.commit();
-
- if(_managedObject!=null)
- {
- _managedObject.unregister();
- }
-
for (Task task : _deleteTaskList)
{
task.doTask(this);
@@ -2167,16 +2147,13 @@ public class SimpleAMQQueue implements A
}
else
{
- if (_managedObject != null)
+ // There is a chance that the node could be deleted by
+ // the time the check actually occurs. So verify we
+ // can actually get the message to perform the check.
+ ServerMessage msg = node.getMessage();
+ if (msg != null)
{
- // There is a chance that the node could be deleted by
- // the time the check actually occurs. So verify we
- // can actually get the message to perform the check.
- ServerMessage msg = node.getMessage();
- if (msg != null)
- {
- _managedObject.checkForNotification(msg);
- }
+ checkForNotification(msg);
}
}
}
@@ -2301,11 +2278,6 @@ public class SimpleAMQQueue implements A
return _notificationChecks;
}
- public ManagedObject getManagedObject()
- {
- return _managedObject;
- }
-
private final class QueueEntryListener implements QueueEntry.StateChangeListener
{
@@ -2520,4 +2492,34 @@ public class SimpleAMQQueue implements A
_maximumDeliveryCount = maximumDeliveryCount;
}
+ /**
+ * Checks if there is any notification to send to the listeners
+ */
+ private void checkForNotification(ServerMessage<?> msg) throws AMQException
+ {
+ final Set<NotificationCheck> notificationChecks = getNotificationChecks();
+ final AMQQueue.NotificationListener listener = _notificationListener;
+
+ if(listener != null && !notificationChecks.isEmpty())
+ {
+ final long currentTime = System.currentTimeMillis();
+ final long thresholdTime = currentTime - getMinimumAlertRepeatGap();
+
+ for (NotificationCheck check : notificationChecks)
+ {
+ if (check.isMessageSpecific() || (_lastNotificationTimes[check.ordinal()] < thresholdTime))
+ {
+ if (check.notifyIfNecessary(msg, this, listener))
+ {
+ _lastNotificationTimes[check.ordinal()] = currentTime;
+ }
+ }
+ }
+ }
+ }
+
+ public void setNotificationListener(AMQQueue.NotificationListener listener)
+ {
+ _notificationListener = listener;
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Sun Jun 10 17:38:18 2012
@@ -62,8 +62,6 @@ public interface VirtualHost extends Dur
void close();
- ManagedObject getManagedObject();
-
UUID getBrokerId();
void scheduleHouseKeepingTask(long period, HouseKeepingTask task);
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java Sun Jun 10 17:38:18 2012
@@ -53,8 +53,6 @@ import org.apache.qpid.server.federation
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.VirtualHostMessages;
import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
-import org.apache.qpid.server.management.ManagedObject;
-import org.apache.qpid.server.management.VirtualHostMBean;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.v1_0.LinkRegistry;
@@ -97,8 +95,6 @@ public class VirtualHostImpl implements
private final BrokerConfig _brokerConfig;
private final VirtualHostConfiguration _vhostConfig;
-
- private final VirtualHostMBean _virtualHostMBean;
private final QueueRegistry _queueRegistry;
@@ -143,8 +139,6 @@ public class VirtualHostImpl implements
CurrentActor.get().message(VirtualHostMessages.CREATED(_name));
- _virtualHostMBean = new VirtualHostMBean(this);
-
_securityManager = new SecurityManager(_appRegistry.getSecurityManager());
_securityManager.configureHostPlugins(_vhostConfig);
@@ -530,11 +524,6 @@ public class VirtualHostImpl implements
CurrentActor.get().message(VirtualHostMessages.CLOSED());
}
- public ManagedObject getManagedObject()
- {
- return _virtualHostMBean;
- }
-
public UUID getBrokerId()
{
return _appRegistry.getBrokerId();
Modified: qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java Sun Jun 10 17:38:18 2012
@@ -650,4 +650,9 @@ public class MockAMQQueue implements AMQ
public void visit(final QueueEntryVisitor visitor)
{
}
+
+ @Override
+ public void setNotificationListener(NotificationListener listener)
+ {
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java Sun Jun 10 17:38:18 2012
@@ -63,7 +63,10 @@ public interface ManagedQueue
String ENCODING = "Encoding";
String CONTENT = "Content";
List<String> VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC = Collections.unmodifiableList(Arrays.asList(MSG_AMQ_ID, MIME, ENCODING, CONTENT));
-
+
+ /** Date/time format used for message expiration and message timestamp formatting */
+ String JMSTIMESTAMP_DATETIME_FORMAT = "MM-dd-yy HH:mm:ss.SSS z";
+
//Individual attribute name constants
static final String ATTR_NAME = "Name";
static final String ATTR_OWNER = "Owner";
Modified: qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagedQueueMBeanTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagedQueueMBeanTest.java?rev=1348629&r1=1348628&r2=1348629&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagedQueueMBeanTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/ManagedQueueMBeanTest.java Sun Jun 10 17:38:18 2012
@@ -24,7 +24,6 @@ import org.apache.log4j.Logger;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.configuration.ClientProperties;
import org.apache.qpid.management.common.mbeans.ManagedQueue;
-import org.apache.qpid.server.management.AMQQueueMBean;
import org.apache.qpid.test.utils.JMXTestUtils;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
@@ -129,7 +128,7 @@ public class ManagedQueueMBeanTest exten
final Map<String, String> headers = headerArrayToMap(headerArray);
final String expectedJMSMessageID = isBroker010() ? sentMessage.getJMSMessageID().replace("ID:", "") : sentMessage.getJMSMessageID();
- final String expectedFormattedJMSTimestamp = FastDateFormat.getInstance(AMQQueueMBean.JMSTIMESTAMP_DATETIME_FORMAT).format(sentMessage.getJMSTimestamp());
+ final String expectedFormattedJMSTimestamp = FastDateFormat.getInstance(ManagedQueue.JMSTIMESTAMP_DATETIME_FORMAT).format(sentMessage.getJMSTimestamp());
assertEquals("Unexpected JMSMessageID within header", expectedJMSMessageID, headers.get("JMSMessageID"));
assertEquals("Unexpected JMSPriority within header", String.valueOf(sentMessage.getJMSPriority()), headers.get("JMSPriority"));
assertEquals("Unexpected JMSTimestamp within header", expectedFormattedJMSTimestamp, headers.get("JMSTimestamp"));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org