diff options
Diffstat (limited to 'java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java')
| -rw-r--r-- | java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java | 29 |
1 files changed, 14 insertions, 15 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java index f734edcc7b..353a2007c0 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java @@ -26,7 +26,6 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.BasicContentHeaderProperties; import org.apache.qpid.framing.ContentBody; import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.FieldTable; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.management.AMQManagedObject; import org.apache.qpid.server.management.MBeanConstructor; @@ -142,8 +141,8 @@ public class AMQQueue implements Managable, Comparable // OpenMBean data types for viewMessageContent method private CompositeType _msgContentType = null; - private String[] _msgContentAttributes = {"Message Id", "MimeType", "Encoding", "Content"}; - private OpenType[] _msgContentAttributeTypes = new OpenType[4]; + private String[] _msgContentAttributes = {"Message Id", "MimeType", "Encoding", "Content"}; + private OpenType[] _msgContentAttributeTypes = new OpenType[4]; @MBeanConstructor("Creates an MBean exposing an AMQQueue") public AMQQueueMBean() throws JMException @@ -163,14 +162,14 @@ public class AMQQueue implements Managable, Comparable _msgContentAttributeTypes[2] = SimpleType.STRING; // For Encoding _msgContentAttributeTypes[3] = new ArrayType(1, SimpleType.BYTE); // For message content _msgContentType = new CompositeType("Message Content", "AMQ Message Content", _msgContentAttributes, - _msgContentAttributes, _msgContentAttributeTypes); + _msgContentAttributes, _msgContentAttributeTypes); _msgAttributeTypes[0] = SimpleType.LONG; // For message id _msgAttributeTypes[1] = new ArrayType(1, SimpleType.STRING); // For header attributes _msgAttributeTypes[2] = SimpleType.LONG; // For size _msgAttributeTypes[3] = SimpleType.BOOLEAN; // For redelivered - _messageDataType = new CompositeType("Message", "AMQ Message", _msgAttributeNames, _msgAttributeNames, _msgAttributeTypes); + _messageDataType = new CompositeType("Message","AMQ Message", _msgAttributeNames, _msgAttributeNames, _msgAttributeTypes); _messagelistDataType = new TabularType("Messages", "List of messages", _messageDataType, _msgAttributeIndex); } @@ -266,7 +265,7 @@ public class AMQQueue implements Managable, Comparable { queueDepth = queueDepth + getMessageSize(message); } - return (long) Math.round(queueDepth / 1000); + return (long)Math.round(queueDepth / 1000); } /** @@ -315,7 +314,7 @@ public class AMQQueue implements Managable, Comparable private void notifyClients(String notificationMsg) { Notification n = new Notification(MonitorNotification.THRESHOLD_VALUE_EXCEEDED, this, - ++_notificationSequenceNumber, System.currentTimeMillis(), notificationMsg); + ++_notificationSequenceNumber, System.currentTimeMillis(), notificationMsg); _broadcaster.sendNotification(n); } @@ -362,7 +361,7 @@ public class AMQQueue implements Managable, Comparable if (msg == null) { - throw new JMException("AMQMessage with message id = " + msgId + " is not in the " + _queueName); + throw new JMException("AMQMessage with message id = " + msgId + " is not in the " + _queueName ); } // get message content List<ContentBody> cBodies = msg.getContentBodies(); @@ -380,7 +379,7 @@ public class AMQQueue implements Managable, Comparable } // Create header attributes list - BasicContentHeaderProperties headerProperties = (BasicContentHeaderProperties) msg.getContentHeaderBody().properties; + BasicContentHeaderProperties headerProperties = (BasicContentHeaderProperties)msg.getContentHeaderBody().properties; String mimeType = headerProperties.getContentType(); String encoding = headerProperties.getEncoding() == null ? "" : headerProperties.getEncoding(); Object[] itemValues = {msgId, mimeType, encoding, msgContent.toArray(new Byte[0])}; @@ -403,12 +402,12 @@ public class AMQQueue implements Managable, Comparable TabularDataSupport _messageList = new TabularDataSupport(_messagelistDataType); // Create the tabular list of message header contents - for (int i = beginIndex; i <= endIndex && i <= list.size(); i++) + for (int i = beginIndex; i <= endIndex && i <= list.size(); i++) { AMQMessage msg = list.get(i - 1); - ContentHeaderBody headerBody = msg.getContentHeaderBody(); + ContentHeaderBody headerBody = msg.getContentHeaderBody(); // Create header attributes list - BasicContentHeaderProperties headerProperties = (BasicContentHeaderProperties) headerBody.properties; + BasicContentHeaderProperties headerProperties = (BasicContentHeaderProperties)headerBody.properties; List<String> headerAttribsList = new ArrayList<String>(); headerAttribsList.add("App Id=" + headerProperties.getAppId()); headerAttribsList.add("MimeType=" + headerProperties.getContentType()); @@ -431,7 +430,7 @@ public class AMQQueue implements Managable, Comparable @Override public MBeanNotificationInfo[] getNotificationInfo() { - String[] notificationTypes = new String[]{MonitorNotification.THRESHOLD_VALUE_EXCEEDED}; + 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); @@ -582,12 +581,12 @@ public class AMQQueue implements Managable, Comparable _bindings.addBinding(routingKey, exchange); } - public void registerProtocolSession(AMQProtocolSession ps, int channel, String consumerTag, boolean acks, FieldTable filters) + public void registerProtocolSession(AMQProtocolSession ps, int channel, String consumerTag, boolean acks) throws AMQException { debug("Registering protocol session {0} with channel {1} and consumer tag {2} with {3}", ps, channel, consumerTag, this); - Subscription subscription = _subscriptionFactory.createSubscription(channel, ps, consumerTag, acks, filters); + Subscription subscription = _subscriptionFactory.createSubscription(channel, ps, consumerTag, acks); _subscribers.addSubscriber(subscription); } |