JMS Distributed Topic

To understand the JMS Concepts in detail we have every concept one experiment that will gives more clarity on each. So here it is about one of the JMS destination type that is 'Distributed Topic'.

Prerequisites

1. WebLogic installed
2. Domain configured
3. JMS Servers configured and module

Current Configurations we will do the following:

• New/Existing JMS Module - with appropriate subdeployment ( pinned or distributed)
• Create New Connection Factory
• Create New Distributed Topic

JMS Topic on WebLogic Server

TopicSend.java

package jms.test;

import java.io.*;
import java.util.*;
import javax.transaction.*;
import javax.naming.*;
import javax.jms.*;
import javax.rmi.PortableRemoteObject;

public class TopicSend
{
 public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";
 public final static String JMS_FACTORY="TestCF";
 public final static String TOPIC="TestTopic1";

 protected TopicConnectionFactory tconFactory;
 protected TopicConnection tcon;
 protected TopicSession tsession;
 protected TopicPublisher tpublisher;
 protected Topic topic;
 protected TextMessage msg;

 public void init(Context ctx, String topicName) throws NamingException, JMSException
 {
  tconFactory = (TopicConnectionFactory) PortableRemoteObject.narrow(ctx.lookup(JMS_FACTORY),TopicConnectionFactory.class);
  tcon = tconFactory.createTopicConnection();
  tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
  topic = (Topic) PortableRemoteObject.narrow(ctx.lookup(topicName), Topic.class);
  tpublisher = tsession.createPublisher(topic);
  msg = tsession.createTextMessage();
  tcon.start();
 }

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

 public void close() throws JMSException {
  tpublisher.close();
  tsession.close();
  tcon.close();
 }

 public static void main(String[] args) throws Exception {
  if (args.length != 1) {
  System.out.println("Usage: java  TopicSend WebLogicURL");
  return;
  }
  InitialContext ic = getInitialContext(args[0]);
  TopicSend ts = new TopicSend();
  ts.init(ic, TOPIC);
  readAndSend(ts);
  ts.close();
 }
 protected static void readAndSend(TopicSend ts)throws IOException, JMSException
 {
  BufferedReader msgStream = new BufferedReader (new InputStreamReader(System.in));
  String line=null;
  System.out.print("\n\t TopicSender Started ... Enter message (\"quit\" to quit): \n");
  do {
  System.out.print("Topic Sender Says > ");
  line = msgStream.readLine();
  if (line != null && line.trim().length() != 0) {
  ts.send(line);
  }
  } while (line != null && ! line.equalsIgnoreCase("quit"));
 }

 protected 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);
  env.put("weblogic.jndi.createIntermediateContexts", "true");
  return new InitialContext(env);
 }
}

TopicReceive.java

package jms.test;

import java.io.*;
import java.util.*;
import javax.transaction.*;
import javax.naming.*;
import javax.jms.*;
import javax.rmi.PortableRemoteObject;
public class TopicReceive implements MessageListener 
 {
 public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";
 public final static String JMS_FACTORY="TestCF";
 public final static String TOPIC="TestTopic1";
 private TopicConnectionFactory tconFactory;
 private TopicConnection tcon;
 private TopicSession tsession;
 private TopicSubscriber tsubscriber;
 private Topic topic;
 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("JMS Message Received: "+ msgText );
if (msgText.equalsIgnoreCase("quit")) {
synchronized(this) {
quit = true;
this.notifyAll();
}
}
} catch (JMSException jmse) {
System.err.println("An exception occurred: "+jmse.getMessage());
}
}

public void init(Context ctx, String topicName)throws NamingException, JMSException
{
tconFactory = (TopicConnectionFactory)PortableRemoteObject.narrow(ctx.lookup(JMS_FACTORY), TopicConnectionFactory.class);
tcon = tconFactory.createTopicConnection();
tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (Topic) PortableRemoteObject.narrow(ctx.lookup(topicName), Topic.class);
tsubscriber = tsession.createSubscriber(topic);
tsubscriber.setMessageListener(this);
tcon.start();
}

public void close() throws JMSException {
tsubscriber.close();
tsession.close();
tcon.close();
}

public static void main(String[] args) throws Exception {
if (args.length != 1) {
System.out.println("Usage:topic.TopicReceive WebLogicURL");
return;
}
InitialContext ic = getInitialContext(args[0]);
TopicReceive tr = new TopicReceive();
tr.init(ic, TOPIC);
System.out.println("JMS Ready To Receive Messages (To quit, send a \"quit\" message).");
synchronized(tr) {
while (! tr.quit) {
try {
tr.wait();
} catch (InterruptedException ie) {}
}
}
tr.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);
env.put("weblogic.jndi.createIntermediateContexts", "true");
return new InitialContext(env);
}
}

Output

Troubleshooting JMS Topic JNDI missing

vagrant@precise64:/vagrant$ java jms.test.TopicSend t3://192.168.33.100:7011,192.168.33.100:7012
Exception in thread "main" javax.naming.NameNotFoundException: Unable to resolve 'TestCF'. Resolved '' [Root exception is javax.naming.NameNotFoundException: Unable to resolve 'TestCF'. Resolved '']; remaining name 'TestCF'
        at weblogic.rjvm.ResponseImpl.unmarshalReturn(ResponseImpl.java:237)
        at weblogic.rmi.cluster.ClusterableRemoteRef.invoke(ClusterableRemoteRef.java:348)
        at weblogic.rmi.cluster.ClusterableRemoteRef.invoke(ClusterableRemoteRef.java:259)
        at weblogic.jndi.internal.ServerNamingNode_1036_WLStub.lookup(Unknown Source)
        at weblogic.jndi.internal.WLContextImpl.lookup(WLContextImpl.java:424)
        at weblogic.jndi.internal.WLContextImpl.lookup(WLContextImpl.java:412)
        at javax.naming.InitialContext.lookup(InitialContext.java:417)
        at jms.test.TopicSend.init(TopicSend.java:25)
        at jms.test.TopicSend.main(TopicSend.java:52)
Caused by: javax.naming.NameNotFoundException: Unable to resolve 'TestCF'. Resolved ''
        at weblogic.jndi.internal.BasicNamingNode.newNameNotFoundException(BasicNamingNode.java:1139)

Fix: Restart entire Weblogic server, if JNDI Tree doesn't show's TestCF