Tag: JMS

Working example of Expiry Queue in JBoss AS 5.1

In this article we would see how and why a expiry queue is given in JBoss. So let’s answer the why first, expiry queue is given because some messages has an expiry period too (i.e. when setTimeToLive is been used) now JMS specification states that clients should not receive messages that have been expired however it does not grantee that. Hence JBoss messaging makes user that once the messages gets expiries are been moved to the expiry queue which has been assigned to the given queue. Now these “expired” messages can later be consumed from the expiry destination for further inspection.

Make sure you do not confuse Expiry Queue with Dead Letter Queue in JBoss because they both have a specific jobs, where DLQ comes into picture when a message does not get delivered in specified delivery attempts (i.e. MaxDeliveryAttempts) to get more information on DLQ you can have a look at the below article

Working example of Dead Letter Queue in JBoss

Features To Discuss

  1. We would be using the default Expiry Queue which comes with JBoss
  2. In this example we will send 1 message with a short time-to-live to a queue.
  3. We will wait for the message to expire and checks that the message is no longer in the queue it was sent to.

Working example of Expiry Queue in JBoss

  1. Create a queue from the Admin-console or even create “*-service.xml” file and place it in the “/profile/deploy” directory. Where * means your Queue name as shown in the below example

    TestQueue-service.xml

  2. <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
    <server>
        <mbean xmbean-dd="xmdesc/Queue-xmbean.xml" name="jboss.messaging.destination:service=Queue,name=TestQueue" code="org.jboss.jms.server.destination.QueueService">
    	<!-- Defining the ExpiryQueue for this TestQueue -->
            <attribute name="ExpiryQueue">jboss.messaging.destination:name=ExpiryQueue,service=Queue</attribute>
            <attribute name="JNDIName">TestQueue</attribute>
            <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends>
            <depends>jboss.messaging:service=PostOffice</depends>
        </mbean>
    </server>
    
  3. That’s it !!! You are done configuring a simple queue with ExpiryQueue, now let’s test this out using the sender and receiver Java code.

Testing

  1. Create a Directory somewhere in your file system like: “/urs/JBoss/JBM/Queue” to write the QueueSend.java and QueueReceive.java programs.
  2. In QueueSend.java copy the below program, where we are setting the first messages to live for 1s (1000ms) before expiration
    import java.io.*;
    import java.io.*;
    import java.util.Hashtable;
    import javax.jms.JMSException;
    import javax.jms.Queue;
    import javax.jms.QueueConnection;
    import javax.jms.QueueConnectionFactory;
    import javax.jms.QueueSender;
    import javax.jms.QueueSession;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.naming.Context;
    import javax.naming.InitialContext;
    import javax.naming.NamingException;
    
    public class QueueSend
    {
    public final static String JNDI_FACTORY="org.jnp.interfaces.NamingContextFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String JMS_FACTORY="/ConnectionFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String QUEUE="/TestQueue";
    
    
    private QueueConnectionFactory qconFactory;
    private QueueConnection qcon;
    private QueueSession qsession;
    private QueueSender qsender;
    private Queue queue;
    private TextMessage msg;
    
    public void init(Context ctx, String queueName)throws NamingException, JMSException
    {
    qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
    qcon = qconFactory.createQueueConnection();
    qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    queue = (Queue) ctx.lookup(queueName);
    qsender = qsession.createSender(queue);
    
    //Messages sent by this sender will be live for 1s (1000ms) before expiration
    qsender.setTimeToLive(1000);
    
    msg = qsession.createTextMessage();
    qcon.start();
    }
    
    public void send(String message) throws JMSException {
    msg.setText(message);
    qsender.send(msg);
    System.out.println("Message="+message+ " Time to live ="+qsender.getTimeToLive());
    }
    
    public void close() throws JMSException {
    qsender.close();
    qsession.close();
    qcon.close();
    }
    
    public static void main(String[] args) throws Exception {
    if (args.length != 1) {
    System.out.println("Usage: java QueueSend URL");
    return;
    }
    InitialContext ic = getInitialContext(args[0]);
    QueueSend qs = new QueueSend();
    qs.init(ic, QUEUE);
    readAndSend(qs);
    qs.close();
    }
    
    private static void readAndSend(QueueSend qs) throws IOException, JMSException
    {
    String line="Test Message Body with counter = ";
    BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
    boolean readFlag=true;
    System.out.println("ntStart Sending Messages (Enter QUIT to Stop):n");
    while(readFlag)
    {
    System.out.print("<Msg_Sender> ");
    String msg=br.readLine();
    if(msg.equals("QUIT") || msg.equals("quit"))
    {
    qs.send(msg);
    System.exit(0);
    }
    qs.send(msg);
    System.out.println();
    }
    br.close();
    }
    
    private static InitialContext getInitialContext(String url) throws NamingException
    {
    Hashtable env = new Hashtable();
    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
    env.put(Context.PROVIDER_URL, url);
    return new InitialContext(env);
    }
    }
    
  3. In QueueReceive.java copy the below program, if you notice that before we are creating the Receiver we are making sure that the program starts 5 seconds late so that the first message sent gets expired as its time to live is 1 second. So the first messages is never consumed and that messages is been moved to expiry queue. But other messages would get consumed as the consumer is up and listening to the TestQueue
    import java.util.Hashtable;
    import javax.jms.*;
    import javax.naming.Context;
    import javax.naming.InitialContext;
    import javax.naming.NamingException;
    
    public class QueueReceive implements MessageListener
    {
    public final static String JNDI_FACTORY="org.jnp.interfaces.NamingContextFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String JMS_FACTORY="/ConnectionFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String QUEUE="/TestQueue";
    
    
    private QueueConnectionFactory qconFactory;
    private QueueConnection qcon;
    private QueueSession qsession;
    private QueueReceiver qreceiver;
    private Queue queue;
    private boolean quit = false;
    
    public void onMessage(Message msg) 
    {
    	try 
    	{
    		String msgText;
    		if (msg instanceof TextMessage)
    		{
    			msgText = ((TextMessage)msg).getText();
    		}
    		else
    		{
    			msgText = msg.toString();
    		}
    
    	 	
    
    		System.out.println("nt<Msg_Receiver> "+ msgText );
    
    		if (msgText.equalsIgnoreCase("quit"))
    		{
    			synchronized(this)
    			{
    				quit = true;
    				this.notifyAll(); // Notify main thread to quit
    			}
    		}
    	}
    	catch (JMSException jmse)
    	{
    		jmse.printStackTrace();
    	}
    }
    
    public void init(Context ctx, String queueName) throws NamingException, JMSException
    {
    	try 
    	{
    		qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
    		qcon = qconFactory.createQueueConnection();
    		qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    		queue = (Queue) ctx.lookup(queueName);
    
    		//Sleeping for 5's so that the message gets expired.
    		System.out.println("Sleeping for 5's till the message expire...");
    		Thread.sleep(5000);
    		
    		qreceiver = qsession.createReceiver(queue);
    		qreceiver.setMessageListener(this);
    		qcon.start();
    	}
    	catch (InterruptedException jmse)
    	{
    		jmse.printStackTrace();
    	}
    
    }
    
    public void close()throws JMSException
    {
    qreceiver.close();
    qsession.close();
    qcon.close();
    }
    
    public static void main(String[] args) throws Exception
    {
    if (args.length != 1)
    {
    System.out.println("Usage: java QueueReceive URL");
    return;
    }
    InitialContext ic = getInitialContext(args[0]);
    QueueReceive qr = new QueueReceive();
    qr.init(ic, QUEUE);
    System.out.println("JMS Ready To Receive Messages (To quit, send a "quit" message from QueueSender.class).");
    
    synchronized(qr)
    {
    while (! qr.quit)
    {
    try
    {
    qr.wait();
    }
    catch (InterruptedException ie)
    {}
    }
    }
    qr.close();
    }
    
    private static InitialContext getInitialContext(String url) throws NamingException
    {
    Hashtable env = new Hashtable();
    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
    env.put(Context.PROVIDER_URL, url);
    return new InitialContext(env);
    }
    }
    
  4. Now open 2– command prompt and run the “. ./setJBossEnv.sh” by adding two DOTs separated by a single space, in both the prompts to set the Environment (PATH & CLASSPATH).
  5. setJBossEnv.sh

    #!/bin/sh
    echo "Exporting JBOSS_HOME...."
    export JBOSS_HOME=/Jboss/AS5.1/jboss-as   <---- Change as per your environment
    echo "JBOSS_HOME Exported....!!!"
    echo ""
    echo "Exporting JAVA_HOME...."
    export JAVA_HOME=/Jdk/jdk1.6.0_21    <---- Change as per your environment
    echo "JBOSS_HOME Exported....!!!"
    echo ""
    echo "Exporting CLASSPATH...."
    export CLASSPATH=$JBOSS_HOME/client/jbossall-client.jar:$JBOSS_HOME/client/jboss-javaee.jar:$JBOSS_HOME/common/lib/jnpserver.jar:$JBOSS_HOME/client/jboss-logging-spi.jar:$JBOSS_HOME/lib/jboss-javaee.jar:$JBOSS_HOME/common/lib/servlet-api.jar:$JBOSS_HOME/common/lib/log4j.jar:$CLASSPATH:.:
    echo "CLASSPATH Exported....!!!"
    echo ""
    echo "Exporting PATH...."
    export PATH=$JAVA_HOME/bin:$PATH
    echo "PATH Exported....!!!"
    echo ""
    echo "========================================================================"
    echo "JBOSS_HOME = " $JBOSS_HOME
    echo ""
    echo "JAVA_HOME = " $JAVA_HOME
    echo ""
    echo "CLASSPATH = " $CLASSPATH
    echo ""
    echo "PATH = " $PATH
    echo "========================================================================"
    
  6. Then compile the QueueSend.java and QueueReceive.java programs.
  7. Now run the code with the following command
  8. Prompt-1

    java QueueSend jnp://localhost:1099
     
  9. Now send one message when the QueueReceive is not started, this way the message which has sent to the TestQueue would have the Time To Live as 1’s second. But this time would not start till the QueueReceive has been startred
  10. Now start the QueueReceive using the below command
  11. Prompt-2

    java QueueReceive jnp://localhost:1099
     

    Note:
    With the “java QueueSend” and “java QueueReceive” command you have to give the<Protocol>://<IP-address>:<Port number>of the managed servers in different prompts.

  12. Once the QueueReceive has been started for the first time it would sleep for 5 second, hence the message which was in the TestQueue would get expired as its time to live was 1 second, hence the messages would get moved to expire queue, which can be seen in the console under
    Admin console path:
    Resources – JBoss Messaging – ExpiryQueue – “Message Count”
  13. However other messages would get consumed as the consumer is up and listening to the TestQueue

Working example of Dead Letter Queue in JBoss AS 5.1

There are times when some messages are delivered unsuccessfully like if the transacted session used to consume them has rolled back for some reason. Such a message goes back to the JMS destination and gets ready to be redelivered as per the configuration settings. It can be possible for a message to be delivered again and again without any success and those messages are never delivered and remain in the destination. These types of messages can become POISON messages, due to which other messages in the queue cannot get consumed and a plie of messages stacks over the queue and result many other issues.

So the question comes into the picture how to prevent this? Here comes the Dead Letter Queue (DLQ) to the rescue, after a specified unsuccessful delivery attempts (i.e. MaxDeliveryAttempts), the message is removed from the destination and moves it inside the defined Dead Letter Queue (DLQ) where the messages can be consumed for further investigation.

Poison Message


Dead Letter Queue (DLQ)


Features To Discuss

  1. We would be using the default DLQ which comes with JBoss
  2. In this example would make sure that if a message is having a string as “Hello World”, then that message would be rolled backed after specified unsuccessful delivery attempts and would be moved in the DLQ and other messages can be consumed without any issue.

Working example of Dead Letter Queue in JBoss

  1. Create a queue from the Admin-console or even create “*-service.xml” file and place it in the “/profile/deploy” directory. Where * means your Queue name as shown in the below example

    TestQueue-service.xml

  2. <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
    <server>
        <mbean xmbean-dd="xmdesc/Queue-xmbean.xml" name="jboss.messaging.destination:service=Queue,name=TestQueue" code="org.jboss.jms.server.destination.QueueService">
    	<!-- Defining the DLQ for this TestQueue -->
            <attribute name="DLQ"> jboss.messaging.destination:name=DLQ,service=Queue</attribute>
            <attribute name="JNDIName">TestQueue</attribute>
    	<!-- The number of times, undilivered messages would be resend -->
    	<attribute name="MaxDeliveryAttempts">3</attribute>
            <depends optional-attribute-name="ServerPeer">jboss.messaging:service=ServerPeer</depends>
            <depends>jboss.messaging:service=PostOffice</depends>
        </mbean>
    </server>
    
  3. That’s it !!! You are done configuring a simple queue having a MaxDeliveryAttempts as 3 with DLQ, now if a message does not get delivered in 3 attempts then that message would be moved to DLQ. Let’s test this out using the sender and receiver Java code.

Testing

  1. Create a Directory somewhere in your file system like: “/urs/JBoss/JBM/Queue” to write the QueueSend.java and QueueReceive.java programs.
  2. In QueueSend.java copy the below program
    import java.io.*;
    import java.io.*;
    import java.util.Hashtable;
    import javax.jms.JMSException;
    import javax.jms.Queue;
    import javax.jms.QueueConnection;
    import javax.jms.QueueConnectionFactory;
    import javax.jms.QueueSender;
    import javax.jms.QueueSession;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.naming.Context;
    import javax.naming.InitialContext;
    import javax.naming.NamingException;
    
    public class QueueSend
    {
    public final static String JNDI_FACTORY="org.jnp.interfaces.NamingContextFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String JMS_FACTORY="/ConnectionFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String QUEUE="/TestQueue";
    
    
    private QueueConnectionFactory qconFactory;
    private QueueConnection qcon;
    private QueueSession qsession;
    private QueueSender qsender;
    private Queue queue;
    private TextMessage msg;
    
    public void init(Context ctx, String queueName)throws NamingException, JMSException
    {
    qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
    qcon = qconFactory.createQueueConnection();
    qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    queue = (Queue) ctx.lookup(queueName);
    qsender = qsession.createSender(queue);
    msg = qsession.createTextMessage();
    qcon.start();
    }
    
    public void send(String message) throws JMSException {
    msg.setText(message);
    qsender.send(msg);
    }
    
    public void close() throws JMSException {
    qsender.close();
    qsession.close();
    qcon.close();
    }
    
    public static void main(String[] args) throws Exception {
    if (args.length != 1) {
    System.out.println("Usage: java QueueSend URL");
    return;
    }
    InitialContext ic = getInitialContext(args[0]);
    QueueSend qs = new QueueSend();
    qs.init(ic, QUEUE);
    readAndSend(qs);
    qs.close();
    }
    
    private static void readAndSend(QueueSend qs) throws IOException, JMSException
    {
    String line="Test Message Body with counter = ";
    BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
    boolean readFlag=true;
    System.out.println("ntStart Sending Messages (Enter QUIT to Stop):n");
    while(readFlag)
    {
    System.out.print("<Msg_Sender> ");
    String msg=br.readLine();
    if(msg.equals("QUIT") || msg.equals("quit"))
    {
    qs.send(msg);
    System.exit(0);
    }
    qs.send(msg);
    System.out.println();
    }
    br.close();
    }
    
    private static InitialContext getInitialContext(String url) throws NamingException
    {
    Hashtable env = new Hashtable();
    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
    env.put(Context.PROVIDER_URL, url);
    return new InitialContext(env);
    }
    }
    
  3. In QueueReceive.java copy the below program, which has a creating a transacted JMS Session and in “onMessage” we are checking it the text message which is been received contents “hello world” ignoring the case would get roll-backed if not would get committed to test our DLQ.
    import java.util.Hashtable;
    import javax.jms.*;
    import javax.naming.Context;
    import javax.naming.InitialContext;
    import javax.naming.NamingException;
    
    public class QueueReceive implements MessageListener
    {
    public final static String JNDI_FACTORY="org.jnp.interfaces.NamingContextFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String JMS_FACTORY="/ConnectionFactory";
    
    //*************** Connection Factory JNDI name *************************
    public final static String QUEUE="/TestQueue";
    
    
    private QueueConnectionFactory qconFactory;
    private QueueConnection qcon;
    private QueueSession qsession;
    private QueueReceiver qreceiver;
    private Queue queue;
    private boolean quit = false;
    
    public void onMessage(Message msg)
    {
    	try 
    	{
    		String msgText;
    		if (msg instanceof TextMessage)
    		{
    			msgText = ((TextMessage)msg).getText();
    		}
    		else
    		{
    			msgText = msg.toString();
    		}
    		if (msgText.equalsIgnoreCase("hello world"))
    		{
    			System.out.println("nt<Msg_Receiver> "+ msgText +", hence rolling back");
    			// Here we are rolling back the session. 
    			// The "Hello World" message which we received is not committed, hence it's undelivered and goes back to the TestQ
    			qsession.rollback();
    		}
    		else
    		{
    			System.out.println("nt<Msg_Receiver> "+ msgText );
    			// Here we are committing the session to acknowledge that we have received the message from the TestQ
    			qsession.commit();
    		}
    			
    		if (msgText.equalsIgnoreCase("quit"))
    		{
    			synchronized(this)
    			{
    				quit = true;
    				this.notifyAll();
    			}
    		}
    	}
    	catch (JMSException jmse)
    	{
    	jmse.printStackTrace();
    	}
    }
    
    
    public void init(Context ctx, String queueName) throws NamingException, JMSException
    {
    	qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
    	qcon = qconFactory.createQueueConnection();
    
    	//Creating a *transacted* JMS Session to test our DLQ
    	qsession = qcon.createQueueSession(true, 0);	
    
    	queue = (Queue) ctx.lookup(queueName);
    	qreceiver = qsession.createReceiver(queue);
    	qreceiver.setMessageListener(this);
    	qcon.start();	
    }
    
    public void close()throws JMSException
    {
    qreceiver.close();
    qsession.close();
    qcon.close();
    }
    
    public static void main(String[] args) throws Exception
    {
    if (args.length != 1)
    {
    System.out.println("Usage: java QueueReceive URL");
    return;
    }
    InitialContext ic = getInitialContext(args[0]);
    QueueReceive qr = new QueueReceive();
    qr.init(ic, QUEUE);
    System.out.println("JMS Ready To Receive Messages (To quit, send a "quit" message from QueueSender.class).");
    
    synchronized(qr)
    {
    while (! qr.quit)
    {
    try
    {
    qr.wait();
    }
    catch (InterruptedException ie)
    {}
    }
    }
    qr.close();
    }
    
    private static InitialContext getInitialContext(String url) throws NamingException
    {
    Hashtable env = new Hashtable();
    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
    env.put(Context.PROVIDER_URL, url);
    return new InitialContext(env);
    }
    }
    
  4. Now open 2– command prompt and run the “. ./setJBossEnv.sh” by adding two DOTs separated by a single space, in both the prompts to set the Environment (PATH & CLASSPATH).
  5. setJBossEnv.sh

    #!/bin/sh
    echo "Exporting JBOSS_HOME...."
    export JBOSS_HOME=/Jboss/AS5.1/jboss-as   <---- Change as per your environment
    echo "JBOSS_HOME Exported....!!!"
    echo ""
    echo "Exporting JAVA_HOME...."
    export JAVA_HOME=/Jdk/jdk1.6.0_21    <---- Change as per your environment
    echo "JBOSS_HOME Exported....!!!"
    echo ""
    echo "Exporting CLASSPATH...."
    export CLASSPATH=$JBOSS_HOME/client/jbossall-client.jar:$JBOSS_HOME/client/jboss-javaee.jar:$JBOSS_HOME/common/lib/jnpserver.jar:$JBOSS_HOME/client/jboss-logging-spi.jar:$JBOSS_HOME/lib/jboss-javaee.jar:$JBOSS_HOME/common/lib/servlet-api.jar:$JBOSS_HOME/common/lib/log4j.jar:$CLASSPATH:.:
    echo "CLASSPATH Exported....!!!"
    echo ""
    echo "Exporting PATH...."
    export PATH=$JAVA_HOME/bin:$PATH
    echo "PATH Exported....!!!"
    echo ""
    echo "========================================================================"
    echo "JBOSS_HOME = " $JBOSS_HOME
    echo ""
    echo "JAVA_HOME = " $JAVA_HOME
    echo ""
    echo "CLASSPATH = " $CLASSPATH
    echo ""
    echo "PATH = " $PATH
    echo "========================================================================"
    
  6. Then compile the QueueSend.java and QueueReceive.java programs.
  7. Now run the code with the following command
  8. Prompt-1

    java QueueSend jnp://localhost:1099
     

    Prompt-2

    java QueueReceive jnp://localhost:1099
     

    Note:
    With the “java QueueSend” and “java QueueReceive” command you have to give the<Protocol>://<IP-address>:<Port number>of the managed servers in different prompts.

  9. Once this is been done try to send some messages (i.e. with “hello world” one of them ) in the Prompt-1 which can be revived on Prompt-2. , however the message which is “hellow world” would be rollbacked and as per the “MaxDeliveryAttempts” value 3 times it would be re-sent and then it would be moved to the DLQ, which can be seen in the console under
    Admin console path:
    Resources – JBoss Messaging – DLQ – “Message Count”


Listing & Deleting Messages from JMS Queue using JMX

Hi,

Many times due to different requirement it is desired to list all the Messages present inside the JMS Queue. Also some times it is required to delete all the messages present inside the JMS Queue. jmx-console can be used to perform the same tasks, but most of the time in production environments we disable the jmx-console or admin-console. So in such scenarios we can use the following kind of JMX code to list and if required then delete the Messages present inside the JMS Queue.

In this sample we will be using the default JMS Queue provided by JBoss as “queue/DLQ” but if you want to create your own JMS Queue then you can refer to the following article “How to create a Queue in JBoss ?“. Also we will be using a simple “QueueSend.java” program to send some test messages in the JMS Queue and then with the help of “ManageJMSQueue.java” program we will list the JMS Messages present inside the JMS Queue and if required we will delete them through the same JMX Code.

Step1). Write the Simple “QueueSend.java” program somewhere in your file system to send some JMS Messages to the JMS Queue “queue/DLQ” as following:

import java.io.*;
import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class QueueSend
{
	public final static String JNDI_FACTORY="org.jnp.interfaces.NamingContextFactory";
	public final static String JMS_FACTORY="/XAConnectionFactory";
	public final static String QUEUE="/queue/DLQ";
	private QueueConnectionFactory qconFactory;
	private QueueConnection qcon;
	private QueueSession qsession;
	private QueueSender qsender;
	private Queue queue;
	private TextMessage msg;

public void init(Context ctx, String queueName)throws NamingException, JMSException
     {
	qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
	qcon = qconFactory.createQueueConnection();
	qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
	queue = (Queue) ctx.lookup(queueName);
	qsender = qsession.createSender(queue);
	msg = qsession.createTextMessage();
	qcon.start();
     }

public void send(String message) throws JMSException
     {
	msg.setText(message);
	qsender.send(msg);
     }

public void close() throws JMSException
     {
	qsender.close();
	qsession.close();
	qcon.close();
     }

public static void main(String[] args) throws Exception {
	if (args.length != 1) {
	System.out.println("Usage: java QueueSend JBossURL");
	System.out.println("Example: java QueueSend localhost:1099");
	return;
	}
	InitialContext ic = getInitialContext(args[0]);
	QueueSend qs = new QueueSend();
	qs.init(ic, QUEUE);
	readAndSend(qs);
	qs.close();
	}

private static void readAndSend(QueueSend qs) throws IOException, JMSException
    {
	String line="Test Message Body with counter = ";
	BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
	boolean readFlag=true;
	System.out.println("ntStart Sending Messages (Enter QUIT to Stop):n");
        int i=1;
	while(readFlag)
	{
	   System.out.print("<Msg_Sender> ");
	   String msg=br.readLine();
	   if(msg.equals("QUIT") || msg.equals("quit"))
	    {
	        qs.send(msg);
        	System.exit(0);
 	    }
        qs.send(msg);
        System.out.println("ttt msg sent: "+i+"t "+msg);
        i++;
        System.out.println();
	}
	br.close();
     }

private static InitialContext getInitialContext(String url) throws NamingException
    {
	Hashtable<String,String> env = new Hashtable<String,String>();
	env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
	env.put(Context.PROVIDER_URL, url);
	return new InitialContext(env);
    }
}

Step2). Now write the “ManageJMSQueue.java” program as following in order to list the messages present inside the JMS Queue “queue/DLQ” and if required delete them through the same code.

import javax.management.*;
import java.io.*;
import java.util.*;
import java.rmi.*;
import javax.naming.*;
import java.io.*;
public class ManageJMSQueue
   {
      private MBeanServerConnection server=null;
      public ManageJMSQueue()
      {
            try{
            Hashtable<String,String> ht=new Hashtable<String,String>();
            ht.put(Context.INITIAL_CONTEXT_FACTORY,"org.jboss.security.jndi.JndiLoginInitialContextFactory");
            ht.put(Context.PROVIDER_URL,"localhost:1099");
            ht.put(Context.SECURITY_PRINCIPAL,"admin");
            ht.put(Context.SECURITY_CREDENTIALS,"admin");
            System.out.println("nt 1- Gotting InitialContext...... ");
            Context ctx = new InitialContext(ht);
            System.out.println("nt 2- Got InitialContext: "+ctx);
            server = (MBeanServerConnection) ctx.lookup("jmx/invoker/RMIAdaptor");
            }
            catch(Exception e)
            {
                System.out.println("nnt Exception inside ManageJMSQueue..."+e);
            }
      }

      public void monitorJMS() throws Exception
      {
           ObjectName objectName=new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue");
           System.out.println("nnServerPeer = "+ (javax.management.ObjectName)server.getAttribute(objectName, new String("ServerPeer")));
           System.out.println("QueueName = "+ (String)server.getAttribute(new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue"), new String("Name")));
           System.out.println("JNDI Name = "+ (String)server.getAttribute(new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue"), new String("JNDIName")));
           System.out.println("FullSize = "+ (Integer)server.getAttribute(new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue"), new String("FullSize")));
      }

     public void listAllJMS_Messages() throws Exception
      {
           ObjectName objectName=new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue");
           List<org.jboss.jms.message.JBossTextMessage> messages=(List<org.jboss.jms.message.JBossTextMessage>)server.invoke(objectName, "listAllMessages" , null, null);
           int count=0;
           for(org.jboss.jms.message.JBossTextMessage msg : messages)
                    System.out.println((++count)+"t"+msg.getText());
      }

     public void removeAllJMS_Messages() throws Exception
      {
           String queueName=(String)server.getAttribute(new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue"), new String("Name"));
           System.out.println("nt Removing all JMS Messages from Queue: "+queueName);
           server.invoke(new ObjectName("jboss.messaging.destination:name=DLQ,service=Queue"), "removeAllMessages" , null, null);
           System.out.println("nt All the Messages are removed from JMS Queue: "+queueName);
      }

     public static void main(String ar[]) throws Exception
       {
            ManageJMSQueue ref=new ManageJMSQueue();
            ref.monitorJMS();
            System.out.println("nt Following Messages Are present inside the JMS Queue:");
            ref.listAllJMS_Messages();
            BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
            System.out.print("nn Please Specify (yes/no) to delete all the messages from JMS Queue ? ");
            String answer="";
            if((answer=br.readLine()).equals("yes"))
                             ref.removeAllJMS_Messages();
            br.close();
       }
  }

Step3). Compile and then run the “QueueSend.java” program as following and then send some messages to the JMS Queue:

javac QueueSend.java
java QueueSend localhost:1099
	Start Sending Messages (Enter QUIT to Stop):

<Msg_Sender> TestMessageOne
			 msg sent: 1	 TestMessageOne

<Msg_Sender> TestMessageTwo
			 msg sent: 2	 TestMessageTwo

<Msg_Sender> TestMessageThree
			 msg sent: 3	 TestMessageThree

<Msg_Sender> quit

Step4). After sending some Messages to the JMS Queue “queue/DLQ” Compile and then run the “ManageJMSQueue” program as following to list the messages present inside the JMS Queue:

javac ManageJMSQueue.java
java ManageJMSQueue localhost:1099

	 1- Gotting InitialContext......
	 2- Got InitialContext: javax.naming.InitialContext@74341960

ServerPeer = jboss.messaging:service=ServerPeer
QueueName = DLQ
JNDI Name = /queue/DLQ
FullSize = 200000

	 Following Messages Are present inside the JMS Queue:
1	TestMessageOne
2	TestMessageTwo
3	TestMessageThree
4	quit

 Please Specify (yes/no) toi delete all the messages from JMS Queue ? yes
	 Removing all JMS Messages from Queue: DLQ
	 All the Messages are removed from JMS Queue: DLQ

.
.
Thanks
Middleware Magic Team


Copyright © 2010-2012 Middleware Magic. All rights reserved. |