Integrating ActiveMQ 5.1 with Tomcat 6

Step 1: Add resource entry to CATALINA_HOME/conf/server.xml under GlobalNamingResources..

    <Resource 
name="jms/ConnectionFactory" 
auth="Container" 
type="org.apache.activemq.ActiveMQConnectionFactory" 
description="JMS Connection Factory" 
factory="org.apache.activemq.jndi.JNDIReferenceFactory" 
brokerURL="tcp://localhost:61616" 
brokerName="LocalActiveMQBroker" 
useEmbeddedBroker="false" 
    /> 
    <Resource 
name="jms/Data" 
auth="Container" 
type="org.apache.activemq.command.ActiveMQTopic" 
description="Data Topic" 
factory="org.apache.activemq.jndi.JNDIReferenceFactory" 
physicalName="queue.data"
    />

Step 2: Add resource link entry to CATALINA_HOME/conf/context.xml (For God knows What reason, An entry the context.xml in META-INF of my webapp did not work)

<ResourceLink global="jms/ConnectionFactory" name="jms/ConnectionFactory" type="javax.jms.ConnectionFactory"/> 
<ResourceLink global="jms/Data" name="jms/Data" type="javax.jms.Topic"/>

Step 3: Add resource ref entry in web.xml

    <resource-ref> 
        <res-ref-name>jms/ConnectionFactory</res-ref-name> 
        <res-type>javax.jms.ConnectionFactory</res-type> 
        <res-auth>Container</res-auth> 
        <res-sharing-scope>Shareable</res-sharing-scope> 
    </resource-ref> 
    <resource-ref> 
        <res-ref-name>jms/Data</res-ref-name> 
        <res-type>javax.jms.Topic</res-type> 
        <res-auth>Container</res-auth> 
        <res-sharing-scope>Shareable</res-sharing-scope> 
    </resource-ref> 

Step 4: Put the following jars in CATALINA_HOME/lib from ActiveMQ_HOME/lib
activemq-core-5.1.0.jar
commons-logging-1.1.jar
geronimo-j2ee-management_1.0_spec-1.0.jar
geronimo-jms_1.1_spec-1.1.1.jar
geronimo-jta_1.0.1B_spec-1.0.1.jar

Step 5: Start the server, you should be able to use JNDI to connect to ActiveMQ.

Step 6: When you use IntialContext in your code, first get the context of "java:comp/env".

    Context initCtx = new InitialContext(); 
    jndiContext = (Context) initCtx.lookup("java:comp/env"); 
    Then you can use: 
    connectionFactory = (ConnectionFactory)jndiContext.lookup("jms/ConnectionFactory"); 
    destination = (Destination)jndiContext.lookup("jms/Data"); 

Then, carry on with normal JMS flow.

JMS Tutorial : Topic Subscriber Client

This is in continuation of my tutorial on JMS. The below client uses topic to communicate with the ActiveMQ.

The jndi.properties

# START SNIPPET: jndi 

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory 

# use the following property to configure the default connector 
java.naming.provider.url = tcp://localhost:61616 

# use the following property to specify the JNDI name the connection factory 
# should appear as. 
#connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry 
connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry 

# register some queues in JNDI using the form 
# queue.[jndiName] = [physicalName] 

# register some topics in JNDI using the form 
# topic.[jndiName] = [physicalName] 
#topic.MyTopic = example.MyTopic 
topic.sample.data = sample.data 

# END SNIPPET: jndi 

/** 
* @author shreyas.purohit 
* 
*/ 
public class SampleJMSConsumer implements Runnable { 

    public void run() { 
        Context jndiContext = null; 
        TopicConnectionFactory connectionFactory = null; 
        TopicConnection connection = null; 
        TopicSession session = null; 
        TopicSubscriber consumer = null; 
        Topic destination = null; 
        String sourceName = null; 
        final int numMsgs; 
        sourceName = "sample.data"; 

        /* 
         * Create a JNDI API InitialContext object 
         */ 
        try { 
            jndiContext = new InitialContext(); 
        } catch (NamingException e) { 
            e.printStackTrace(); 
            System.exit(1); 
        } 

        /* 
         * Look up connection factory and destination. 
         */ 
        try { 
            connectionFactory = (TopicConnectionFactory) jndiContext 
                    .lookup("topicConnectionFactry"); 
            destination = (Topic)jndiContext.lookup(sourceName); 
        } catch (NamingException e) { 
            e.printStackTrace(); 
            System.exit(1); 
        } 

        try { 
            connection = connectionFactory.createTopicConnection(); 
            session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE); 
            consumer = session.createSubscriber(destination); 
            connection.start(); 
            try { 
                Thread.sleep(2000); 
            } catch (InterruptedException e) { 
                // TODO Auto-generated catch block 
                e.printStackTrace(); 
            } 
            MessageListener listener = new MyTopicMessageListener(); 
            consumer.setMessageListener(listener); 
            // Let the thread run for some time so that the Consumer has 
            // suffcient time to consume the message 
            try { 
                Thread.sleep(50000); 
            } catch (InterruptedException e) { 
                // TODO Auto-generated catch block 
                e.printStackTrace(); 
            } 
        } catch (JMSException e) { 
            e.printStackTrace(); 
        } finally { 
            if (connection != null) { 
                try { 
                    connection.close(); 
                } catch (JMSException e) { 
                } 
            } 
        } 
    } 

} 

The message listener:

/** 
* @author shreyas.purohit 
* 
*/ 
public class MyTopicMessageListener implements MessageListener { 

    /* (non-Javadoc) 
     * @see javax.jms.MessageListener#onMessage(javax.jms.Message) 
     */ 
    public void onMessage(Message arg0) { 
        if(arg0 instanceof ObjectMessage){ 
            try { 
            //Print it out 
            System.out.println("Recieved message in listener: " + ((ObjectMessage)arg0).getObject()); 
            System.out.println("Co-Rel Id: " + ((ObjectMessage)arg0).getJMSCorrelationID()); 
            }catch(Exception e){ 
                e.printStackTrace(); 
                System.exit(1); 
            } 
        }else{ 
            System.out.println("~~~~Error in format~~~"); 
        } 
    } 

} 

The JMS Application:

/** 
* @author shreyas.purohit 
* 
*/ 
public class JMSApp { 

    /** 
     * @param args 
     */ 
    public static void main(String[] args) { 
        runInNewthread(new SampleJMSConsumer()); 
    } 
    public static void runInNewthread(Runnable runnable) { 
        Thread brokerThread = new Thread(runnable); 
        brokerThread.setDaemon(false); 
        brokerThread.start(); 
    } 
}

RMI Tutorial : RMI and tomcat

Using RMI in java is very very easy, but, people like me will still get stuck using it, when trying to integrate with Tomcat. And, the help is very less outside in the internet.

There are two things to remember:
1. Provide sufficient access priviledges to the classes and jars using the RMI. This is done using catalina.policy present in $CATALINA_HOME/conf. Below is the addition to the policy files:

grant codeBase "file:${catalina.home}/webapps/MyAPP/WEB-INF/classes/-" { 
    permission java.security.AllPermission "", ""; 
}; 

grant codeBase "file:${catalina.home}/webapps/MyAPP/WEB-INF/lib/-" { 
    permission java.security.AllPermission "", ""; 
}; 
grant codeBase "file:${catalina.home}/webapps/MyAPP/WEB-INF/lib/some-common-3.0.jar" { 
    permission java.io.FilePermission "*", "read, write"; 
}; 

2. Do not copy the entire code that is avalable on the internet for the server and client. The main() method given at lot of places in the web should not be copied as it is. Specifically, comment out the below snippet on both client and server:

//        if (System.getSecurityManager() == null) { 
//            System.setSecurityManager(new RMISecurityManager()); 
//        } 

We do not want to install a new security manager. Tomcat provides the securitymanager, and lets use the same. Please do use the same if you want the RMI to work.
Tutorial:

Server side:

Step 1: Create an contract, an interface that extends java.rmi.Remote.

public interface IRemoteService extends Remote{ 

    public final String serviceName = "MyRemoteService"; 
    public abstract void startDoing() throws RemoteException; 

    public abstract void stopDoing() throws RemoteException; 
} 

Step 2: Write an implementation for the Interface.

public class RemoteServiceImpl implements IRemoteService { 
    public RemoteServiceImpl(){ 
        super(); 
    } 
    public void startDoing() throws RemoteException { 
        return new MyTask().do(); 
    } 
    public void stopDoing() throws RemoteException { 
        return new MyTask().dont(); 
    }    
} 

Step 3: Either use a startup servlet or any class that is called after the server is up and running and before the remote service is invoked, and initialize the registry. In the blow snippet case, just create a new object some where before service will be invoked. Note in below code, there is no securitymanager present anywhere.

public class InitRemoteService { 
    public static boolean isRegistered = false; 
    public static IRemoteService service; 
    public InitRemoteService(){ 
     if(!isRegistered){ 
        try { 
            service = new RemoteServiceImpl(); 
            IRemoteService stub = 
                (IRemoteService) UnicastRemoteObject.exportObject(service, 0); 
            Registry registry = LocateRegistry.createRegistry(9345); 
            registry.rebind(IRemoteService.serviceName, stub); 
            System.out.println("Remote service bound"); 
            isRegistered = true; 
        } catch (Exception e) { 
            System.err.println("Remote service exception:"); 
            e.printStackTrace(); 
        } 
    } 
    } 
} 

Client Side:

Step 1: Write the client as given below in the snippet. You can note, there is no security manager related code either here. It also, lists all the service names in the registry.

try { 
    Registry registry = LocateRegistry.getRegistry(HOST,9345); 
    String[] names = registry.list(); 
    for(String name1 : names){ 
        System.out.println("~~~~" + name1 + "~~~~"); 
    } 
    IDPRemoteService serv = (IDPRemoteService) registry.lookup(IDPRemoteService.serviceName); 
    System.out.println(serv.startDoing()); 
} catch (Exception e) { 
    System.err.println("Remoteservice exception:"); 
    e.printStackTrace(); 
}

Enjoy, using RMI for remote object invocations, esp on servers like Tomcat that does not support EJB's or when EJB level of advanced concept is not required.

Accessing resources from Java Class

Here is a simple code snippet that can be used to get properties file from a java program.

Properties config = new Properties(); 
try { 
    config.load(new FileInputStream(new File(URLDecoder.decode(getClass().getClassLoader().getResource(CONFIG_FILE_NAME).getFile(), "UTF-8")))); 
} catch (Exception e1) { 
    e1.printStackTrace(); 
    throw new RuntimeException(e1); 
} 

This, snippet has never let me down till now in getting a properties file.

ObjectOutputStream : Writing the same object over and over again!

Well, Well, Well.. You know, I was just writing a very tiny socket application in Java and was stuck for a long time on 'Writing Objects from server to Client!'. I myself can not believe that. With so many samples around on the Internet, I was still stuck. A pretty interesting problem I would like to share. I have a server writing objects to its clients. But the client is always receiving the same object. The first one. The code snippet is given below:

....
....
....
os = new ObjectOutputStream(socket.getOutputStream());
....
....
....

if(toBeSentData.shouldConsume()){
    Object object = toBeSentData.get();
    if(null != object){
        os.writeObject(object);
        os.flush();
    }
}

You can see, I am flushing the output stream. I debugged through the writeObject code, to find out that my Object was not written at all. Some handle was written. Specifically, in method private void writeObject0(Object obj, boolean unshared) throws IOException, the snippet is present below.

// handle previously written and non-replaceable objects
int h;
if ((obj = subs.lookup(obj)) == null) {
    writeNull();
    return;
} else if (!unshared && (h = handles.lookup(obj)) != -1) {
    writeHandle(h);
    return;
} else if (obj instanceof Class) {
    writeClass((Class) obj, unshared);
    return;
} else if (obj instanceof ObjectStreamClass) {
    writeClassDesc((ObjectStreamClass) obj, unshared);
    return;
}

The handles.lookup(obj) was never returning -1. So, the problem is some sort of caching(Not exactly caching) I understood. The only guy who had explained this was Qusay H. Mahmoud in his December 2001 post about Advanced Socket Programming at http://java.sun.com/developer/technicalArticles/ALT/sockets/
I really thank him for that, else I would have been stuck with the problemo for god knows How long!!

I will just copy the last part of the article here. This is what gave me the solution.

Object Serialization Pitfall

When working with object serialization it is important to keep in mind that the ObjectOutputStream maintains a hashtable mapping the objects written into the stream to a handle. When an object is written to the stream for the first time, its contents will be copied to the stream. Subsequent writes, however, result in a handle to the object being written to the stream. This may lead to a couple of problems:

    * If an object is written to the stream then modified and written a second time, the modifications will not be noticed when the stream is deserialized. Again, the reason is that subsequent writes results in the handle being written but the modified object is not copied into the stream. To solve this problem, call the ObjectOutputStream.reset method that discards the memory of having sent an object so subsequent writes copy the object into the stream.
    * An OutOfMemoryError may be thrown after writing a large number of objects into the ObjectOutputStream. The reason for this is that the hashtable maintains references to objects that might otherwise be unreachable by an application. This problem can be solved simply by calling the ObjectOutputStream.reset method to reset the object/handle table to its initial state. After this call, all previously written objects will be eligible for garbage collection.

The reset method resets the stream state to be the same as if it had just been constructed. This method may not be called while objects are being serialized. Inappropriate invocations of this method result in an IOException.

Just call the os.reset(); after flush. It worked like a charm for me!!