activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a..@apache.org
Subject svn commit: r379400 - in /incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console: command/ filter/ util/
Date Tue, 21 Feb 2006 09:29:32 GMT
Author: aco
Date: Tue Feb 21 01:29:32 2006
New Revision: 379400

URL: http://svn.apache.org/viewcvs?rev=379400&view=rev
Log:
Added initial implementation for JMX queue browsing and purging. They're still unable to filter messages. I'm still not sure if we should use MessageSelector or filter them in the client. Currently, browsing still uses the old AMQ client implementation.

Added:
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/AmqBrowseCommand.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/PurgeCommand.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/AmqMessagesQueryFilter.java
Modified:
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/BrowseCommand.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/ShellCommand.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MBeansRegExQueryFilter.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MapTransformFilter.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MessagesQueryFilter.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/AmqMessagesUtil.java
    incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/JmxMBeansUtil.java

Added: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/AmqBrowseCommand.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/AmqBrowseCommand.java?rev=379400&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/AmqBrowseCommand.java (added)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/AmqBrowseCommand.java Tue Feb 21 01:29:32 2006
@@ -0,0 +1,240 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.console.command;
+
+import org.apache.activemq.console.util.AmqMessagesUtil;
+import org.apache.activemq.console.formatter.GlobalWriter;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+
+import javax.jms.Destination;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.StringTokenizer;
+import java.util.Iterator;
+
+public class AmqBrowseCommand extends AbstractAmqCommand {
+    public static final String QUEUE_PREFIX = "queue:";
+    public static final String TOPIC_PREFIX = "topic:";
+
+    public static final String VIEW_GROUP_HEADER = "header:";
+    public static final String VIEW_GROUP_CUSTOM = "custom:";
+    public static final String VIEW_GROUP_BODY   = "body:";
+
+    private final List queryAddObjects = new ArrayList(10);
+    private final List querySubObjects = new ArrayList(10);
+    private final Set  groupViews      = new HashSet(10);
+    private final Set  queryViews      = new HashSet(10);
+
+    /**
+     * Execute the browse command, which allows you to browse the messages in a given JMS destination
+     * @param tokens - command arguments
+     * @throws Exception
+     */
+    protected void runTask(List tokens) throws Exception {
+        try {
+            // If no destination specified
+            if (tokens.isEmpty()) {
+                GlobalWriter.printException(new IllegalArgumentException("No JMS destination specified."));
+                return;
+            }
+
+            // If no broker url specified
+            if (getBrokerUrl() == null) {
+                GlobalWriter.printException(new IllegalStateException("No broker url specified. Use the --amqurl option to specify a broker url."));
+                return;
+            }
+
+            // Display the messages for each destination
+            for (Iterator i=tokens.iterator(); i.hasNext();) {
+                String destName = (String)i.next();
+                Destination dest;
+
+                // If destination has been explicitly specified as a queue
+                if (destName.startsWith(QUEUE_PREFIX)) {
+                    dest = new ActiveMQQueue(destName.substring(QUEUE_PREFIX.length()));
+
+                // If destination has been explicitly specified as a topic
+                } else if (destName.startsWith(TOPIC_PREFIX)) {
+                    dest = new ActiveMQTopic(destName.substring(TOPIC_PREFIX.length()));
+
+                // By default destination is assumed to be a queue
+                } else {
+                    dest = new ActiveMQQueue(destName);
+                }
+
+                // Query for the messages to view
+                List addMsgs = AmqMessagesUtil.getMessages(getBrokerUrl(), dest, queryAddObjects);
+
+                // Query for the messages to remove from view
+                if (querySubObjects.size() > 0) {
+                    List subMsgs = AmqMessagesUtil.getMessages(getBrokerUrl(), dest, querySubObjects);
+                    addMsgs.removeAll(subMsgs);
+                }
+
+                // Display the messages
+                GlobalWriter.printMessage(AmqMessagesUtil.filterMessagesView(addMsgs, groupViews, queryViews));
+            }
+
+        } catch (Exception e) {
+            GlobalWriter.printException(new RuntimeException("Failed to execute browse task. Reason: " + e));
+            throw new Exception(e);
+        }
+    }
+
+    /**
+     * Handle the --msgsel, --xmsgsel, --view, -V options.
+     * @param token - option token to handle
+     * @param tokens - succeeding command arguments
+     * @throws Exception
+     */
+    protected void handleOption(String token, List tokens) throws Exception {
+
+        // If token is an additive message selector option
+        if (token.startsWith("--msgsel")) {
+
+            // If no message selector is specified, or next token is a new option
+            if (tokens.isEmpty() || ((String)tokens.get(0)).startsWith("-")) {
+                GlobalWriter.printException(new IllegalArgumentException("Message selector not specified"));
+                return;
+            }
+
+            StringTokenizer queryTokens = new StringTokenizer((String)tokens.remove(0), COMMAND_OPTION_DELIMETER);
+            while (queryTokens.hasMoreTokens()) {
+                queryAddObjects.add(queryTokens.nextToken());
+            }
+        }
+
+        // If token is a substractive message selector option
+        else if (token.startsWith("--xmsgsel")) {
+
+            // If no message selector is specified, or next token is a new option
+            if (tokens.isEmpty() || ((String)tokens.get(0)).startsWith("-")) {
+                GlobalWriter.printException(new IllegalArgumentException("Message selector not specified"));
+                return;
+            }
+
+            StringTokenizer queryTokens = new StringTokenizer((String)tokens.remove(0), COMMAND_OPTION_DELIMETER);
+            while (queryTokens.hasMoreTokens()) {
+                querySubObjects.add(queryTokens.nextToken());
+            }
+
+        }
+
+        // If token is a view option
+        else if (token.startsWith("--view")) {
+
+            // If no view specified, or next token is a new option
+            if (tokens.isEmpty() || ((String)tokens.get(0)).startsWith("-")) {
+                GlobalWriter.printException(new IllegalArgumentException("Attributes to view not specified"));
+                return;
+            }
+
+            // Add the attributes to view
+            StringTokenizer viewTokens = new StringTokenizer((String)tokens.remove(0), COMMAND_OPTION_DELIMETER);
+            while (viewTokens.hasMoreTokens()) {
+                String viewToken = viewTokens.nextToken();
+
+                // If view is explicitly specified to belong to the JMS header
+                if (viewToken.equals(VIEW_GROUP_HEADER)) {
+                    queryViews.add(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + viewToken.substring(VIEW_GROUP_HEADER.length()));
+
+                // If view is explicitly specified to belong to the JMS custom header
+                } else if (viewToken.equals(VIEW_GROUP_CUSTOM)) {
+                    queryViews.add(AmqMessagesUtil.JMS_MESSAGE_CUSTOM_PREFIX + viewToken.substring(VIEW_GROUP_CUSTOM.length()));
+
+                // If view is explicitly specified to belong to the JMS body
+                } else if (viewToken.equals(VIEW_GROUP_BODY)) {
+                    queryViews.add(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX + viewToken.substring(VIEW_GROUP_BODY.length()));
+
+                // If no view explicitly specified, let's check the view for each group
+                } else {
+                    queryViews.add(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + viewToken);
+                    queryViews.add(AmqMessagesUtil.JMS_MESSAGE_CUSTOM_PREFIX + viewToken);
+                    queryViews.add(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX   + viewToken);
+                }
+            }
+        }
+
+        // If token is a predefined group view option
+        else if (token.startsWith("-V")) {
+            String viewGroup = token.substring(2);
+            // If option is a header group view
+            if (viewGroup.equals("header")) {
+                groupViews.add(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX);
+
+            // If option is a custom header group view
+            } else if (viewGroup.equals("custom")) {
+                groupViews.add(AmqMessagesUtil.JMS_MESSAGE_CUSTOM_PREFIX);
+
+            // If option is a body group view
+            } else if (viewGroup.equals("body")) {
+                groupViews.add(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX);
+
+            // Unknown group view
+            } else {
+                GlobalWriter.printInfo("Unknown group view: " + viewGroup + ". Ignoring group view option.");
+            }
+        }
+
+        // Let super class handle unknown option
+        else {
+            super.handleOption(token, tokens);
+        }
+    }
+
+    /**
+     * Print the help messages for the browse command
+     */
+    protected void printHelp() {
+        GlobalWriter.printHelp(helpFile);
+    }
+
+    protected String[] helpFile = new String[] {
+        "Task Usage: Main browse --amqurl <broker url> [browse-options] <destinations>",
+        "Description: Display selected destination's messages.",
+        "",
+        "Browse Options:",
+        "    --amqurl <url>                Set the broker URL to connect to.",
+        "    --msgsel <msgsel1,msglsel2>   Add to the search list messages matched by the query similar to",
+        "                                  the messages selector format.",
+        "    -V<header|custom|body>        Predefined view that allows you to view the message header, custom",
+        "                                  message header, or the message body.",
+        "    --view <attr1>,<attr2>,...    Select the specific attribute of the message to view.",
+        "    --version                     Display the version information.",
+        "    -h,-?,--help                  Display the browse broker help information.",
+        "",
+        "Examples:",
+        "    Main browse --amqurl tcp://localhost:61616 FOO.BAR",
+        "        - Print the message header, custom message header, and message body of all messages in the",
+        "          queue FOO.BAR",
+        "",
+        "    Main browse --amqurl tcp://localhost:61616 -Vheader,body queue:FOO.BAR",
+        "        - Print only the message header and message body of all messages in the queue FOO.BAR",
+        "",
+        "    Main browse --amqurl tcp://localhost:61616 -Vheader --view custom:MyField queue:FOO.BAR",
+        "        - Print the message header and the custom field 'MyField' of all messages in the queue FOO.BAR",
+        "",
+        "    Main browse --amqurl tcp://localhost:61616 --msgsel JMSMessageID='*:10',JMSPriority>5 FOO.BAR",
+        "        - Print all the message fields that has a JMSMessageID in the header field that matches the",
+        "          wildcard *:10, and has a JMSPriority field > 5 in the queue FOO.BAR",
+        "        * To use wildcard queries, the field must be a string and the query enclosed in ''",
+        "",
+    };
+}

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/BrowseCommand.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/BrowseCommand.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/BrowseCommand.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/BrowseCommand.java Tue Feb 21 01:29:32 2006
@@ -16,20 +16,19 @@
  */
 package org.apache.activemq.console.command;
 
-import org.apache.activemq.console.util.AmqMessagesUtil;
 import org.apache.activemq.console.formatter.GlobalWriter;
-import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.console.util.AmqMessagesUtil;
+import org.apache.activemq.console.util.JmxMBeansUtil;
 
-import javax.jms.Destination;
+import javax.management.ObjectInstance;
 import java.util.List;
+import java.util.StringTokenizer;
 import java.util.ArrayList;
 import java.util.Set;
 import java.util.HashSet;
-import java.util.StringTokenizer;
 import java.util.Iterator;
 
-public class BrowseCommand extends AbstractAmqCommand {
+public class BrowseCommand extends AbstractJmxCommand {
     public static final String QUEUE_PREFIX = "queue:";
     public static final String TOPIC_PREFIX = "topic:";
 
@@ -49,49 +48,21 @@
      */
     protected void runTask(List tokens) throws Exception {
         try {
-            // If no destination specified
+            // If there is no queue name specified, let's select all
             if (tokens.isEmpty()) {
-                GlobalWriter.printException(new IllegalArgumentException("No JMS destination specified."));
-                return;
-            }
-
-            // If no broker url specified
-            if (getBrokerUrl() == null) {
-                GlobalWriter.printException(new IllegalStateException("No broker url specified. Use the --amqurl option to specify a broker url."));
-                return;
+                tokens.add("*");
             }
 
-            // Display the messages for each destination
+            // Iterate through the queue names
             for (Iterator i=tokens.iterator(); i.hasNext();) {
-                String destName = (String)i.next();
-                Destination dest;
+                List queueList = JmxMBeansUtil.queryMBeans(useJmxServiceUrl(), "Type=Queue,Destination=" + i.next() + ",*");
 
-                // If destination has been explicitly specified as a queue
-                if (destName.startsWith(QUEUE_PREFIX)) {
-                    dest = new ActiveMQQueue(destName.substring(QUEUE_PREFIX.length()));
-
-                // If destination has been explicitly specified as a topic
-                } else if (destName.startsWith(TOPIC_PREFIX)) {
-                    dest = new ActiveMQTopic(destName.substring(TOPIC_PREFIX.length()));
-
-                // By default destination is assumed to be a queue
-                } else {
-                    dest = new ActiveMQQueue(destName);
+                // Iterate through the queue result
+                for (Iterator j=queueList.iterator(); j.hasNext();) {
+                    List messages = JmxMBeansUtil.createMessageQueryFilter(useJmxServiceUrl(), ((ObjectInstance)j.next()).getObjectName()).query(queryAddObjects);
+                    GlobalWriter.printMessage(JmxMBeansUtil.filterMessagesView(messages, groupViews, queryViews));
                 }
-
-                // Query for the messages to view
-                List addMsgs = AmqMessagesUtil.getMessages(getBrokerUrl(), dest, queryAddObjects);
-
-                // Query for the messages to remove from view
-                if (querySubObjects.size() > 0) {
-                    List subMsgs = AmqMessagesUtil.getMessages(getBrokerUrl(), dest, querySubObjects);
-                    addMsgs.removeAll(subMsgs);
-                }
-
-                // Display the messages
-                GlobalWriter.printMessage(AmqMessagesUtil.filterMessagesView(addMsgs, groupViews, queryViews));
             }
-
         } catch (Exception e) {
             GlobalWriter.printException(new RuntimeException("Failed to execute browse task. Reason: " + e));
             throw new Exception(e);
@@ -207,31 +178,31 @@
     }
 
     protected String[] helpFile = new String[] {
-        "Task Usage: Main browse --amqurl <broker url> [browse-options] <destinations>",
+        "Task Usage: Main browse [browse-options] <destinations>",
         "Description: Display selected destination's messages.",
         "",
         "Browse Options:",
-        "    --amqurl <url>                Set the broker URL to connect to.",
         "    --msgsel <msgsel1,msglsel2>   Add to the search list messages matched by the query similar to",
         "                                  the messages selector format.",
         "    -V<header|custom|body>        Predefined view that allows you to view the message header, custom",
         "                                  message header, or the message body.",
         "    --view <attr1>,<attr2>,...    Select the specific attribute of the message to view.",
+        "    --jmxurl <url>                Set the JMX URL to connect to.",
         "    --version                     Display the version information.",
         "    -h,-?,--help                  Display the browse broker help information.",
         "",
         "Examples:",
-        "    Main browse --amqurl tcp://localhost:61616 FOO.BAR",
+        "    Main browse FOO.BAR",
         "        - Print the message header, custom message header, and message body of all messages in the",
         "          queue FOO.BAR",
         "",
-        "    Main browse --amqurl tcp://localhost:61616 -Vheader,body queue:FOO.BAR",
+        "    Main browse -Vheader,body queue:FOO.BAR",
         "        - Print only the message header and message body of all messages in the queue FOO.BAR",
         "",
-        "    Main browse --amqurl tcp://localhost:61616 -Vheader --view custom:MyField queue:FOO.BAR",
+        "    Main browse -Vheader --view custom:MyField queue:FOO.BAR",
         "        - Print the message header and the custom field 'MyField' of all messages in the queue FOO.BAR",
         "",
-        "    Main browse --amqurl tcp://localhost:61616 --msgsel JMSMessageID='*:10',JMSPriority>5 FOO.BAR",
+        "    Main browse --msgsel JMSMessageID='*:10',JMSPriority>5 FOO.BAR",
         "        - Print all the message fields that has a JMSMessageID in the header field that matches the",
         "          wildcard *:10, and has a JMSPriority field > 5 in the queue FOO.BAR",
         "        * To use wildcard queries, the field must be a string and the query enclosed in ''",

Added: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/PurgeCommand.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/PurgeCommand.java?rev=379400&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/PurgeCommand.java (added)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/PurgeCommand.java Tue Feb 21 01:29:32 2006
@@ -0,0 +1,176 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.console.command;
+
+import org.apache.activemq.console.formatter.GlobalWriter;
+import org.apache.activemq.console.util.JmxMBeansUtil;
+
+import javax.management.ObjectInstance;
+import javax.management.ObjectName;
+import javax.management.MBeanServerConnection;
+import javax.management.openmbean.CompositeData;
+import javax.management.remote.JMXConnector;
+import java.util.List;
+import java.util.StringTokenizer;
+import java.util.ArrayList;
+import java.util.Iterator;
+
+public class PurgeCommand extends AbstractJmxCommand {
+
+    private final List queryAddObjects = new ArrayList(10);
+    private final List querySubObjects = new ArrayList(10);
+
+    /**
+     * Execute the purge command, which allows you to purge the messages in a given JMS destination
+     * @param tokens - command arguments
+     * @throws Exception
+     */
+    protected void runTask(List tokens) throws Exception {
+        try {
+            // If there is no queue name specified, let's select all
+            if (tokens.isEmpty()) {
+                tokens.add("*");
+            }
+
+            // Iterate through the queue names
+            for (Iterator i=tokens.iterator(); i.hasNext();) {
+                List queueList = JmxMBeansUtil.queryMBeans(useJmxServiceUrl(), "Type=Queue,Destination=" + i.next() + ",*");
+
+                for (Iterator j=queueList.iterator(); j.hasNext();) {
+                    ObjectName queueName = ((ObjectInstance)j.next()).getObjectName();
+                    if (queryAddObjects.isEmpty()) {
+                        purgeQueue(queueName);
+                    } else {
+                        List messages = JmxMBeansUtil.createMessageQueryFilter(useJmxServiceUrl(), queueName).query(queryAddObjects);
+                        purgeMessages(queueName, messages);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            GlobalWriter.printException(new RuntimeException("Failed to execute purge task. Reason: " + e));
+            throw new Exception(e);
+        }
+    }
+
+    /**
+     * Purge all the messages in the queue
+     * @param queue - ObjectName of the queue to purge
+     * @throws Exception
+     */
+    public void purgeQueue(ObjectName queue) throws Exception {
+        JMXConnector conn = createJmxConnector();
+        MBeanServerConnection server = conn.getMBeanServerConnection();
+        GlobalWriter.printInfo("Purging all messages in queue: " + queue.getKeyProperty("Destination"));
+        server.invoke(queue, "purge", new Object[] {}, new String[] {});
+        conn.close();
+    }
+
+    /**
+     * Purge selected messages in the queue
+     * @param queue - ObjectName of the queue to purge the messages from
+     * @param messages - List of messages to purge
+     * @throws Exception
+     */
+    public void purgeMessages(ObjectName queue, List messages) throws Exception {
+        JMXConnector conn = createJmxConnector();
+        MBeanServerConnection server = conn.getMBeanServerConnection();
+
+        Object[] param = new Object[1];
+        for (Iterator i=messages.iterator(); i.hasNext();) {
+            CompositeData msg = (CompositeData)i.next();
+            param[0] = "" + msg.get("JMSMessageID");
+            GlobalWriter.printInfo("Removing message: " + param[0] + " from queue: " + queue.getKeyProperty("Destination"));
+            server.invoke(queue, "removeMessage", param, new String[] {"java.lang.String"});
+        }
+
+        conn.close();
+    }
+
+    /**
+     * Handle the --msgsel, --xmsgsel.
+     * @param token - option token to handle
+     * @param tokens - succeeding command arguments
+     * @throws Exception
+     */
+    protected void handleOption(String token, List tokens) throws Exception {
+        // If token is an additive message selector option
+        if (token.startsWith("--msgsel")) {
+
+            // If no message selector is specified, or next token is a new option
+            if (tokens.isEmpty() || ((String)tokens.get(0)).startsWith("-")) {
+                GlobalWriter.printException(new IllegalArgumentException("Message selector not specified"));
+                return;
+            }
+
+            StringTokenizer queryTokens = new StringTokenizer((String)tokens.remove(0), COMMAND_OPTION_DELIMETER);
+            while (queryTokens.hasMoreTokens()) {
+                queryAddObjects.add(queryTokens.nextToken());
+            }
+        }
+
+        // If token is a substractive message selector option
+        else if (token.startsWith("--xmsgsel")) {
+
+            // If no message selector is specified, or next token is a new option
+            if (tokens.isEmpty() || ((String)tokens.get(0)).startsWith("-")) {
+                GlobalWriter.printException(new IllegalArgumentException("Message selector not specified"));
+                return;
+            }
+
+            StringTokenizer queryTokens = new StringTokenizer((String)tokens.remove(0), COMMAND_OPTION_DELIMETER);
+            while (queryTokens.hasMoreTokens()) {
+                querySubObjects.add(queryTokens.nextToken());
+            }
+
+        }
+
+        // Let super class handle unknown option
+        else {
+            super.handleOption(token, tokens);
+        }
+    }
+
+    /**
+     * Print the help messages for the browse command
+     */
+    protected void printHelp() {
+        GlobalWriter.printHelp(helpFile);
+    }
+
+    protected String[] helpFile = new String[] {
+        "Task Usage: Main purge [browse-options] <destinations>",
+        "Description: Delete selected destination's messages that matches the message selector.",
+        "",
+        "Browse Options:",
+        "    --msgsel <msgsel1,msglsel2>   Add to the search list messages matched by the query similar to",
+        "                                  the messages selector format.",
+        "    --jmxurl <url>                Set the JMX URL to connect to.",
+        "    --version                     Display the version information.",
+        "    -h,-?,--help                  Display the browse broker help information.",
+        "",
+        "Examples:",
+        "    Main purge FOO.BAR",
+        "        - Delete all the messages in queue FOO.BAR",
+
+        "    Main purge --msgsel JMSMessageID='*:10',JMSPriority>5 FOO.*",
+        "        - Delete all the messages in the destinations that matches FOO.* and has a JMSMessageID in",
+        "          the header field that matches the wildcard *:10, and has a JMSPriority field > 5 in the",
+        "          queue FOO.BAR",
+        "        * To use wildcard queries, the field must be a string and the query enclosed in ''",
+        "",
+    };
+}

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/ShellCommand.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/ShellCommand.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/ShellCommand.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/command/ShellCommand.java Tue Feb 21 01:29:32 2006
@@ -69,7 +69,9 @@
             } else if (taskToken.equals("query")) {
                 new QueryCommand().execute(tokens);
             } else if (taskToken.equals("browse")) {
-                new BrowseCommand().execute(tokens);
+                new AmqBrowseCommand().execute(tokens);
+            } else if (taskToken.equals("purge")) {
+                new PurgeCommand().execute(tokens);
             } else {
                 // If not valid task, push back to list
                 tokens.add(0, taskToken);

Added: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/AmqMessagesQueryFilter.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/AmqMessagesQueryFilter.java?rev=379400&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/AmqMessagesQueryFilter.java (added)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/AmqMessagesQueryFilter.java Tue Feb 21 01:29:32 2006
@@ -0,0 +1,152 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.console.filter;
+
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.ActiveMQConnectionFactory;
+
+import javax.jms.Destination;
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Session;
+import javax.jms.QueueBrowser;
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+import java.util.Iterator;
+
+public class AmqMessagesQueryFilter extends AbstractQueryFilter {
+
+    private URI brokerUrl;
+    private Destination destination;
+
+    /**
+     * Create a JMS message query filter
+     * @param brokerUrl - broker url to connect to
+     * @param destination - JMS destination to query
+     */
+    public AmqMessagesQueryFilter(URI brokerUrl, Destination destination) {
+        super(null);
+        this.brokerUrl   = brokerUrl;
+        this.destination = destination;
+    }
+
+    /**
+     * Queries the specified destination using the message selector format query
+     * @param queries - message selector queries
+     * @return list messages that matches the selector
+     * @throws Exception
+     */
+    public List query(List queries) throws Exception {
+        String selector = "";
+
+        // Convert to message selector
+        for (Iterator i=queries.iterator(); i.hasNext();) {
+            selector = selector + "(" + i.next().toString() + ") AND ";
+        }
+
+        // Remove last AND
+        if (selector != "") {
+            selector = selector.substring(0, selector.length() - 5);
+        }
+
+        if (destination instanceof ActiveMQQueue) {
+            return queryMessages((ActiveMQQueue)destination, selector);
+        } else {
+            return queryMessages((ActiveMQTopic)destination, selector);
+        }
+    }
+
+    /**
+     * Query the messages of a queue destination using a queue browser
+     * @param queue - queue destination
+     * @param selector - message selector
+     * @return list of messages that matches the selector
+     * @throws Exception
+     */
+    protected List queryMessages(ActiveMQQueue queue, String selector) throws Exception {
+        Connection conn = createConnection(getBrokerUrl());
+
+        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        QueueBrowser browser = sess.createBrowser(queue, selector);
+
+        List messages = Collections.list(browser.getEnumeration());
+
+        conn.close();
+
+        return messages;
+    }
+
+    /**
+     * Query the messages of a topic destination using a message consumer
+     * @param topic - topic destination
+     * @param selector - message selector
+     * @return list of messages that matches the selector
+     * @throws Exception
+     */
+    protected List queryMessages(ActiveMQTopic topic, String selector) throws Exception {
+        // TODO: should we use a durable subscriber or a retroactive non-durable subscriber?
+        // TODO: if a durable subscriber is used, how do we manage it? subscribe/unsubscribe tasks?
+        return null;
+    }
+
+    /**
+     * Create and start a JMS connection
+     * @param brokerUrl - broker url to connect to.
+     * @return JMS connection
+     * @throws JMSException
+     */
+    protected Connection createConnection(URI brokerUrl) throws JMSException {
+        Connection conn = (new ActiveMQConnectionFactory(brokerUrl)).createConnection();
+        conn.start();
+        return conn;
+    }
+
+    /**
+     * Get the broker url being used.
+     * @return broker url
+     */
+    public URI getBrokerUrl() {
+        return brokerUrl;
+    }
+
+    /**
+     * Set the broker url to use.
+     * @param brokerUrl - broker url
+     */
+    public void setBrokerUrl(URI brokerUrl) {
+        this.brokerUrl = brokerUrl;
+    }
+
+    /**
+     * Get the destination being used.
+     * @return - JMS destination
+     */
+    public Destination getDestination() {
+        return destination;
+    }
+
+    /**
+     * Set the destination to use.
+     * @param destination - JMS destination
+     */
+    public void setDestination(Destination destination) {
+        this.destination = destination;
+    }
+
+}

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MBeansRegExQueryFilter.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MBeansRegExQueryFilter.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MBeansRegExQueryFilter.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MBeansRegExQueryFilter.java Tue Feb 21 01:29:32 2006
@@ -83,7 +83,7 @@
     }
 
     /**
-     * Try to match the attribute list using teh regular expression map
+     * Try to match the attribute list using the regular expression map
      * @param data - attribute list to match
      * @param regex - regex map
      * @return true if the attribute list matches the regex map

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MapTransformFilter.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MapTransformFilter.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MapTransformFilter.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MapTransformFilter.java Tue Feb 21 01:29:32 2006
@@ -30,6 +30,7 @@
 import javax.management.ObjectName;
 import javax.management.AttributeList;
 import javax.management.Attribute;
+import javax.management.openmbean.CompositeDataSupport;
 import javax.jms.JMSException;
 import javax.jms.DeliveryMode;
 import java.util.Map;
@@ -253,6 +254,61 @@
                 props.setProperty(AmqMessagesUtil.JMS_MESSAGE_CUSTOM_PREFIX + name, msg.getObjectProperty(name).toString());
             }
         }
+
+        return props;
+    }
+
+    /**
+     * Transform an openMBean composite data to a Map
+     * @param data - composite data to transform
+     * @return map object
+     */
+    protected Map transformToMap(CompositeDataSupport data) {
+        Properties props = new Properties();
+
+        String typeName = data.getCompositeType().getTypeName();
+
+        // Retrieve text message
+        if (typeName.equals(ActiveMQTextMessage.class.getName())) {
+            props.setProperty(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX + "Text", data.get("Text").toString());
+
+        // Retrieve byte preview
+        } else if (typeName.equals(ActiveMQBytesMessage.class.getName())) {
+            props.setProperty(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX + "BodyLength", data.get("BodyLength").toString());
+            props.setProperty(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX + "BodyPreview", new String((byte[])data.get("BodyPreview")));
+
+        // Expand content map
+        } else if (typeName.equals(ActiveMQMapMessage.class.getName())) {
+            Map contentMap = (Map)data.get("ContentMap");
+            for (Iterator i=contentMap.keySet().iterator(); i.hasNext();) {
+                String key = (String)i.next();
+                props.setProperty(AmqMessagesUtil.JMS_MESSAGE_BODY_PREFIX + key, contentMap.get(key).toString());
+            }
+
+        // Do nothing
+        } else if (typeName.equals(ActiveMQObjectMessage.class.getName()) ||
+                   typeName.equals(ActiveMQStreamMessage.class.getName()) ||
+                   typeName.equals(ActiveMQMessage.class.getName())) {
+
+        // Unrecognized composite data. Throw exception.
+        } else {
+            throw new IllegalArgumentException("Unrecognized composite data to transform. composite type: " + typeName);
+        }
+
+        // Process the JMS message header values
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSCorrelationID", "" + data.get("JMSCorrelationID"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSDestination",   "" + data.get("JMSDestination"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSMessageID",     "" + data.get("JMSMessageID"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSReplyTo",       "" + data.get("JMSReplyTo"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSType",          "" + data.get("JMSType"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSDeliveryMode",  "" + data.get("JMSDeliveryMode"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSExpiration",    "" + data.get("JMSExpiration"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSPriority",      "" + data.get("JMSPriority"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSRedelivered",   "" + data.get("JMSRedelivered"));
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_HEADER_PREFIX + "JMSTimestamp",     "" + data.get("JMSTimestamp"));
+
+        // Process the JMS custom message properties
+        props.setProperty(AmqMessagesUtil.JMS_MESSAGE_CUSTOM_PREFIX + "Properties", "" + data.get("Properties"));
 
         return props;
     }

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MessagesQueryFilter.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MessagesQueryFilter.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MessagesQueryFilter.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/filter/MessagesQueryFilter.java Tue Feb 21 01:29:32 2006
@@ -16,34 +16,32 @@
  */
 package org.apache.activemq.console.filter;
 
-import org.apache.activemq.command.ActiveMQTopic;
-import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.activemq.ActiveMQConnectionFactory;
-
-import javax.jms.Destination;
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Session;
-import javax.jms.QueueBrowser;
-import java.net.URI;
-import java.util.Collections;
+import javax.management.remote.JMXConnector;
+import javax.management.remote.JMXServiceURL;
+import javax.management.remote.JMXConnectorFactory;
+import javax.management.MBeanServerConnection;
+import javax.management.ObjectName;
+import javax.management.openmbean.CompositeData;
+import java.net.MalformedURLException;
 import java.util.List;
 import java.util.Iterator;
+import java.util.Arrays;
+import java.io.IOException;
 
 public class MessagesQueryFilter extends AbstractQueryFilter {
 
-    private URI brokerUrl;
-    private Destination destination;
+    private JMXServiceURL jmxServiceUrl;
+    private ObjectName    destName;
 
     /**
      * Create a JMS message query filter
-     * @param brokerUrl - broker url to connect to
-     * @param destination - JMS destination to query
+     * @param jmxServiceUrl - JMX service URL to connect to
+     * @param destName - object name query to retrieve the destination
      */
-    public MessagesQueryFilter(URI brokerUrl, Destination destination) {
+    public MessagesQueryFilter(JMXServiceURL jmxServiceUrl, ObjectName destName) {
         super(null);
-        this.brokerUrl   = brokerUrl;
-        this.destination = destination;
+        this.jmxServiceUrl = jmxServiceUrl;
+        this.destName      = destName;
     }
 
     /**
@@ -65,88 +63,54 @@
             selector = selector.substring(0, selector.length() - 5);
         }
 
-        if (destination instanceof ActiveMQQueue) {
-            return queryMessages((ActiveMQQueue)destination, selector);
-        } else {
-            return queryMessages((ActiveMQTopic)destination, selector);
-        }
-    }
-
-    /**
-     * Query the messages of a queue destination using a queue browser
-     * @param queue - queue destination
-     * @param selector - message selector
-     * @return list of messages that matches the selector
-     * @throws Exception
-     */
-    protected List queryMessages(ActiveMQQueue queue, String selector) throws Exception {
-        Connection conn = createConnection(getBrokerUrl());
-
-        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        QueueBrowser browser = sess.createBrowser(queue, selector);
-
-        List messages = Collections.list(browser.getEnumeration());
-
-        conn.close();
-
-        return messages;
+        return queryMessages(selector);
     }
 
     /**
-     * Query the messages of a topic destination using a message consumer
-     * @param topic - topic destination
+     * Query the messages of a queue destination using JMX
      * @param selector - message selector
      * @return list of messages that matches the selector
      * @throws Exception
      */
-    protected List queryMessages(ActiveMQTopic topic, String selector) throws Exception {
-        // TODO: should we use a durable subscriber or a retroactive non-durable subscriber?
-        // TODO: if a durable subscriber is used, how do we manage it? subscribe/unsubscribe tasks?
-        return null;
-    }
+    protected List queryMessages(String selector) throws Exception {
+        JMXConnector connector = createJmxConnector();
+        MBeanServerConnection server = connector.getMBeanServerConnection();
+        CompositeData[] messages = (CompositeData[])server.invoke(destName, "browse", new Object[] {}, new String[] {});
+        connector.close();
 
-    /**
-     * Create and start a JMS connection
-     * @param brokerUrl - broker url to connect to.
-     * @return JMS connection
-     * @throws JMSException
-     */
-    protected Connection createConnection(URI brokerUrl) throws JMSException {
-        Connection conn = (new ActiveMQConnectionFactory(brokerUrl)).createConnection();
-        conn.start();
-        return conn;
+        return Arrays.asList(messages);
     }
 
     /**
-     * Get the broker url being used.
-     * @return broker url
+     * Get the JMX service URL the query is connecting to.
+     * @return JMX service URL
      */
-    public URI getBrokerUrl() {
-        return brokerUrl;
+    public JMXServiceURL getJmxServiceUrl() {
+        return jmxServiceUrl;
     }
 
     /**
-     * Set the broker url to use.
-     * @param brokerUrl - broker url
+     * Sets the JMX service URL the query is going to connect to.
+     * @param jmxServiceUrl - new JMX service URL
      */
-    public void setBrokerUrl(URI brokerUrl) {
-        this.brokerUrl = brokerUrl;
+    public void setJmxServiceUrl(JMXServiceURL jmxServiceUrl) {
+        this.jmxServiceUrl = jmxServiceUrl;
     }
 
     /**
-     * Get the destination being used.
-     * @return - JMS destination
+     * Sets the JMX service URL the query is going to connect to.
+     * @param jmxServiceUrl - new JMX service URL
      */
-    public Destination getDestination() {
-        return destination;
+    public void setJmxServiceUrl(String jmxServiceUrl) throws MalformedURLException {
+        setJmxServiceUrl(new JMXServiceURL(jmxServiceUrl));
     }
 
     /**
-     * Set the destination to use.
-     * @param destination - JMS destination
+     * Creates a JMX connector
+     * @return JMX connector
+     * @throws java.io.IOException
      */
-    public void setDestination(Destination destination) {
-        this.destination = destination;
+    protected JMXConnector createJmxConnector() throws IOException {
+        return JMXConnectorFactory.connect(getJmxServiceUrl());
     }
-
 }

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/AmqMessagesUtil.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/AmqMessagesUtil.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/AmqMessagesUtil.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/AmqMessagesUtil.java Tue Feb 21 01:29:32 2006
@@ -18,11 +18,11 @@
 
 import org.apache.activemq.console.filter.QueryFilter;
 import org.apache.activemq.console.filter.WildcardToMsgSelectorTransformFilter;
-import org.apache.activemq.console.filter.MessagesQueryFilter;
 import org.apache.activemq.console.filter.PropertiesViewFilter;
 import org.apache.activemq.console.filter.StubQueryFilter;
 import org.apache.activemq.console.filter.MapTransformFilter;
 import org.apache.activemq.console.filter.GroupPropertiesViewFilter;
+import org.apache.activemq.console.filter.AmqMessagesQueryFilter;
 
 import javax.jms.Destination;
 import java.net.URI;
@@ -58,7 +58,7 @@
 
     public static QueryFilter createMessageQueryFilter(URI brokerUrl, Destination dest) {
         return new WildcardToMsgSelectorTransformFilter(
-            new MessagesQueryFilter(brokerUrl, dest)
+            new AmqMessagesQueryFilter(brokerUrl, dest)
         );
     }
 }

Modified: incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/JmxMBeansUtil.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/JmxMBeansUtil.java?rev=379400&r1=379399&r2=379400&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/JmxMBeansUtil.java (original)
+++ incubator/activemq/trunk/activemq-console/src/main/java/org/apache/activemq/console/util/JmxMBeansUtil.java Tue Feb 21 01:29:32 2006
@@ -16,16 +16,10 @@
  */
 package org.apache.activemq.console.util;
 
-import org.apache.activemq.console.filter.QueryFilter;
-import org.apache.activemq.console.filter.MBeansObjectNameQueryFilter;
-import org.apache.activemq.console.filter.WildcardToRegExTransformFilter;
-import org.apache.activemq.console.filter.MBeansRegExQueryFilter;
-import org.apache.activemq.console.filter.MBeansAttributeQueryFilter;
-import org.apache.activemq.console.filter.PropertiesViewFilter;
-import org.apache.activemq.console.filter.StubQueryFilter;
-import org.apache.activemq.console.filter.MapTransformFilter;
+import org.apache.activemq.console.filter.*;
 
 import javax.management.remote.JMXServiceURL;
+import javax.management.ObjectName;
 import java.util.Set;
 import java.util.List;
 import java.util.Iterator;
@@ -112,5 +106,21 @@
                 )
             )
         );
+    }
+
+    public static QueryFilter createMessageQueryFilter(JMXServiceURL jmxUrl, ObjectName destName) {
+        return new WildcardToMsgSelectorTransformFilter(
+            new MessagesQueryFilter(jmxUrl, destName)
+        );
+    }
+
+    public static List filterMessagesView(List messages, Set groupViews, Set attributeViews) throws Exception {
+        return (new PropertiesViewFilter(attributeViews,
+            new GroupPropertiesViewFilter(groupViews,
+                new MapTransformFilter(
+                    new StubQueryFilter(messages)
+                )
+            )
+        )).query("");
     }
 }



Mime
View raw message