read message from Topic

[Vlade Maksimovic]

Hello...
How to read message from Topic without the use of MessageListener...
I try with MessageConsumer receive() method, but.... nothing...

When I try to work with Queue, and when I use receive method, all works fine.

Thanks

[Peter Johnson]

Please post the code you are using, including the code that establishes the connection to the topic.

Topics are different than queues - only listeners who are registered at the time that a message is posted to a topic will received the message. You can change this by registering a durable listener, in which case the server will ensure that the durable listener gets all messages, even those posted while the listener was not listening.

[Vlade Maksimovic]

Producer:

  
package com.pool;

import java.io.IOException;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.XAConnectionFactory;
import javax.management.InstanceNotFoundException;
import javax.management.MBeanException;
import javax.management.ObjectName;
import javax.management.ReflectionException;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NameNotFoundException;
import javax.naming.NamingException;

import org.jboss.jmx.adaptor.rmi.RMIAdaptor;
import org.jboss.mx.util.ObjectNameFactory;

public class Producer {

	private Context contex = null;
	private Properties properties = null;
	private Connection connection = null;
	private Session session = null;
	private MessageProducer publisher = null;
	private Topic receiver = null;
	private TextMessage message = null;

	public Producer(){
		try {
			properties = new Properties();
			properties.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
			properties.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
			properties.setProperty("java.naming.provider.url", "jnp://localhost:1099");
			contex = new InitialContext(properties);
			XAConnectionFactory cf = (XAConnectionFactory) contex.lookup("/XAConnectionFactory");
			connection = cf.createXAConnection();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		} catch (NamingException e) {
			e.printStackTrace();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	public void sendMessage(String topicName, String poruka) throws JMSException, InstanceNotFoundException, MBeanException, ReflectionException, IOException, NamingException{
			
		try{
			contex.lookup("topic/" + topicName);
		}catch (NameNotFoundException e) {
			ObjectName destinationManager = ObjectNameFactory.create("jboss.messaging:service=ServerPeer");
			RMIAdaptor server = null;
			server = (RMIAdaptor) contex.lookup("jmx/invoker/RMIAdaptor");
			server.invoke(destinationManager, "deployTopic", new Object[] { topicName, "topic/" + topicName }, new String[] {
					String.class.getName(), String.class.getName() });
		}
		
		receiver = session.createTopic(topicName);	
		publisher = session.createProducer(receiver);
		message = session.createTextMessage();
		message.setText(poruka);
		publisher.send(message);
		publisher.close();
	}
}

Customer:

 
package com.pool;

import java.util.Properties;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.XAConnectionFactory;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class Consumer {

	private Properties properties = null;
	private Context contex = null;
	private Connection connection = null;
	private Session session = null;
	private Topic receiver = null;
	private int id = 0;

	public Consumer(int i) throws JMSException {

		try {
			id = i;
			properties = new Properties();
			properties.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
			properties.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
			properties.setProperty("java.naming.provider.url", "jnp://localhost:1099");
			contex = new InitialContext(properties);
			XAConnectionFactory cf = (XAConnectionFactory) contex.lookup("/XAConnectionFactory");
			connection = cf.createXAConnection();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			connection.start();
		} catch (NamingException e) {
			e.printStackTrace();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}


	public void getMessage(String topicName) throws JMSException{
		receiver = session.createTopic(topicName);
		TextMessage message;
		MessageConsumer consumer = session.createConsumer(receiver);
		message = (TextMessage) consumer.receive();
		if (message != null){
			System.out.println("Consumer " + id + ": " + message.getText());
		}
	}

}

Test class:

   
package com.pool;

import java.io.IOException;

import javax.jms.JMSException;
import javax.management.InstanceNotFoundException;
import javax.management.MBeanException;
import javax.management.ReflectionException;
import javax.naming.NamingException;

public class Test {

	/**
	 * @param args
	 * @throws JMSException 
	 */
	public static void main(String[] args) throws JMSException {
		// TODO Auto-generated method stub
		Producer p = new Producer();

		try {

			for (int i = 0; i < 10; i++){

				p.sendMessage("testTopic5", "test Message "+i);

			}

			System.out.println("Send"); 

			Consumer[] c = new Consumer[10]; 

			for (int i = 0; i < 10; i++){
				c[i] = new Consumer(i);
				c[i].getMessage("testTopic5");
			}
			


		} catch (InstanceNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MBeanException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ReflectionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (NamingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

I known to do this example to work with MessageListener, but my task is that make to work without Listener. I forget to tell, I work with Jboss 5.0.1GA.

[Peter Johnson]

Please read this sentence again:

Topics are different than queues - only listeners who are registered at the time that a message is posted to a topic will received the message.

Then look at your test code and tell me why it does not work.

Your consumer is a little trickier - you will have to read the javadocs for the MessageConsumer.receive() method before you can tell me why it doesn't work.