JbossMQ 3.2.2 delivers duplicates ... ?!
Hi all,
Any clues as to why JBoss delivering duplicate messages? Am I missing something (important), or is the following test program (modified from the JBoss manual) buggy in some way. It effectively creates a topic connection and publishes three messages into it. Listeners are assigned to the subscriber to get async replies. The following program produces the following output (note the redeliveries - and note that Jboss does not think they are redelivered ...)
onMessage, recv text=First Message : 1070472279032 redelivered: false
onMessage, recv text=Second Message : 1070472279112 redelivered: false
onMessage, recv text=Second Message : 1070472279112 redelivered: false
onMessage, recv text=Third Message : 1070472279123 redelivered: false
onMessage, recv text=Third Message : 1070472279123 redelivered: false
onMessage, recv text=Third Message : 1070472279123 redelivered: false
// ******** Code ******
import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSubscriber;
import javax.jms.TopicSession;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import EDU.oswego.cs.dl.util.concurrent.CountDown;
public class TopicSendRecvClient {
static CountDown done;
TopicConnection conn = null;
TopicSession session = null;
Topic topic = null;
static int counter = 3;
public static class ExListener implements MessageListener {
public void onMessage(Message msg) {
TextMessage tm = (TextMessage) msg;
try {
//msg.acknowledge();
System.out.println(
"onMessage, recv text="
+ tm.getText()
+ " redelivered: "
+ msg.getJMSRedelivered());
} catch (Throwable t) {
t.printStackTrace();
}
counter--;
if (counter == 0)
done.release();
}
}
public void setupPubSub() throws JMSException, NamingException {
Hashtable env = new Hashtable();
env.put(
Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
env.put(
Context.URL_PKG_PREFIXES,
"org.jboss.naming:org.jnp.interfaces");
env.put(Context.PROVIDER_URL, "localhost");
InitialContext iniCtx = new InitialContext(env);
Object tmp = iniCtx.lookup("ConnectionFactory");
TopicConnectionFactory tcf = (TopicConnectionFactory) tmp;
conn = tcf.createTopicConnection();
topic = (Topic) iniCtx.lookup("topic/testTopic");
session = conn.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
conn.start();
}
public void sendRecvAsync(String text)
throws JMSException, NamingException {
// Set the async listener
TopicSubscriber recv = session.createSubscriber(topic);
recv.setMessageListener(new ExListener());
// Send a text msg
TopicPublisher send = session.createPublisher(topic);
TextMessage tm = session.createTextMessage(text);
send.publish(tm);
send.close();
}
public void stop() throws JMSException {
conn.stop();
session.close();
conn.close();
}
public static void main(String args[]) throws Exception {
long startTime = 0;
TopicSendRecvClient client = new TopicSendRecvClient();
client.setupPubSub();
client.done = new CountDown(1);
startTime = System.currentTimeMillis();
client.sendRecvAsync("First Message : " + System.currentTimeMillis());
client.done = new CountDown(1);
startTime = System.currentTimeMillis();
client.sendRecvAsync("Second Message : " + System.currentTimeMillis());
client.done = new CountDown(1);
startTime = System.currentTimeMillis();
client.sendRecvAsync("Third Message : " + System.currentTimeMillis());
client.done.acquire();
System.exit(0);
}
}