Search code examples
javajmsweblogicjmxmbeans

Accessing Weblogic JMS using Java and JMX/MBeans


I am trying to write a Java program that can browse all of the JMS queues in a Weblogic JMS server and read the messages in a given queue (but not consume them). I'm trying to use Weblogic Mbeans and JMX but am new to both. I have the following code to get all of the queues and their depth:

private void countMessages1() throws Exception {
    JMXConnector connector = getMBeanServerConnector("/jndi/"+RuntimeServiceMBean.MBEANSERVER_JNDI_NAME);
    MBeanServerConnection mbeanServerConnection = connector.getMBeanServerConnection();
    ObjectName service = new ObjectName("com.bea:Name=RuntimeService,Type=weblogic.management.mbeanservers.runtime.RuntimeServiceMBean");
    ObjectName serverRuntime = (ObjectName) mbeanServerConnection.getAttribute(service, "ServerRuntime");
    ObjectName jmsRuntime = (ObjectName) mbeanServerConnection.getAttribute(serverRuntime, "JMSRuntime");
    ObjectName[] jmsServers = (ObjectName[]) mbeanServerConnection.getAttribute(jmsRuntime, "JMSServers");
    for (ObjectName jmsServer: jmsServers) {
        if ("JMSServer".equals(jmsServer.getKeyProperty("Name"))) {
            ObjectName[] destinations = (ObjectName[]) mbeanServerConnection.getAttribute(jmsServer, "Destinations");
            for (ObjectName destination: destinations) {
                String queueName = destination.getKeyProperty("Name");
                Long queueDepth = (Long) mbeanServerConnection.getAttribute(destination, "MessagesCurrentCount");
                System.out.println("Queue: " + queueName + " Depth: " + queueDepth);
            }
            break;
        }
    }
    connector.close();
}

I also have the ability to delete everything from a queue:

mbeanServerConnection.invoke(destination, "deleteMessages", new Object[] {""}, new String[] {"java.lang.String"});

Where I'm stuck is how to read the actual messages in a destination/queue. I've been playing around with mbeanServerConnection.invoke and I see getMessage and getMessages but I am not sure how to use them properly. Can someone please show an example of how to use those to browse messages in a destination (but not consume them)? I've tried a few variants like this but I can't get it to work:

String message = (String) mbeanServerConnection.invoke(destination, "getMessage", new Object[] { "", 0, JMS_ALL_STATES}, new String[] {"java.lang.String"});

Solution

  • After some study, I was able to write this code that does what I want:

    for (ObjectName destination: destinations) {
        if (destination.getKeyProperty("Name").equalsIgnoreCase(selectedQueue)) {
            try {
                String cursor = (String)mbeanServerConnection.invoke(destination, "getMessages", new Object[] {"", 0}, new String[] {"java.lang.String", "java.lang.Integer"});
                Long cursorSize = (Long)mbeanServerConnection.invoke(destination, "getCursorSize", new Object[] {cursor}, new String[] {"java.lang.String"});
                //System.out.println(cursor + ": " + cursorSize);
                CompositeData[] messages = (CompositeData[])mbeanServerConnection.invoke(destination, "getNext", new Object[] {cursor, cursorSize.intValue()}, new String[] {"java.lang.String", "java.lang.Integer"});
                if (null != messages) {
                    for (CompositeData message: messages) {
                        JMSMessageInfo messageInfo = new JMSMessageInfo(message);
                        Long messageInfoHandle = messageInfo.getHandle();
                        CompositeData messageCursor = (CompositeData)mbeanServerConnection.invoke(destination, "getMessage", new Object[] {cursor, messageInfoHandle}, new String[] {"java.lang.String", "java.lang.Long"});
                        JMSMessageInfo mbi = new JMSMessageInfo(messageCursor);
                        WLMessage messageBody = mbi.getMessage();
                        body = messageBody.toString().substring(messageBody.toString().indexOf(", ") + 2, messageBody.toString().length()-1);
                        Object[] row = {counter, body};
                        publish(row);
                        if (isCancelled()) {
                            modelMessages.setRowCount(0);
                            return model;
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                modelMessages.setRowCount(0);
                return model;
            }
        }
    }