You are viewing an old version of this page. View the current version.
Compare with Current
View Page History
« Previous
Version 3
Next »
Introduction
This document describes how messages can be sent to and received from a messaging queue and how those messages can be processed in a Java class to add an order to a JobScheduler job chain.
No JobScheduler libraries are needed for this example.
Mode Of Operation
- Creating a message producer to send messages to a Message Queue (MQ) server.
- Creating a message consumer to receive messages from a MQ server.
- Sending the message to a Jobscheduler.
To be able to process the message on the desired JobScheduler the message has to be a XML snippet wich can be processed by the JobScheduler API respectively by the JobScheduler XML interface.
The Message Producer
This example describes how to build a connection to a MQ server and send a message to a queue on this server. This example uses Apache Active MQ as the message queue server.
Most of the implementation was done with the standard jms implementation of Java. The only class from the active MQ implementation is the ActiveMQConnectionFactory class.
- Create a connection.
- Create a session.
- Create a destination.
- Create a producer.
- Send a message with the producer.
- Close the connection.
The Methods
To make the implementation more readable as well as reusable, the steps are separated in different methods.
The createConnection(String url) method
This method instantiates a ConnectionFactory with an ActiveMQConnectionFactory and creates a connection through the factory method createConnection().
The ActiveMQConnectionFactory has to be instantiated with the URL of the MQ server.
private Connection createConnection(String uri){
ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
Connection connection = null;
try {
connection = factory.createConnection();
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to connect: " , e);
}
return connection;
}
The createSession(Connection connection) method
The method is called with an already instantiated connection and starts a session through the connection objects method.
private Session createSession(Connection connection){
Session session = null;
try {
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to create Session: " , e);
}
return session;
}
The createDestination(Session session) method
TBD
package com.sos.jms.producer;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;
public class SOSProducer {
private static final String DEFAULT_CONNECTION_URI = "tcp://[HOSTNAME]:61616";
private static final Logger LOGGER = Logger.getLogger(SOSProducer.class);
private static final String QUEUE_NAME = "test_queue";
private static final String TEXT = "<add_order job_chain='[FOLDERNAME]/[JOBCHAINNAME]' at='now'>"
+ "<params>"
+ "<param name='host' value='[HOST]'/>"
+ "<param name='port' value='22'/>"
+ "<param name='user' value='[USERNAME]'/>"
+ "<param name='auth_method' value='password'/>"
+ "<param name='password' value='[PASSWORD]'/>"
+ "<param name='command' value='echo command send over MessageQueue!'/>"
+ "</params>"
+ "</add_order>";
private Connection createConnection(){
return this.createConnection(DEFAULT_CONNECTION_URI);
}
private Connection createConnection(String uri){
ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
Connection connection = null;
try {
connection = factory.createConnection();
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to connect: " , e);
}
return connection;
}
private Session createSession(Connection connection){
Session session = null;
try {
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to create Session: " , e);
}
return session;
}
private Destination createDestination(Session session){
Destination destination = null;
try {
destination = session.createQueue(QUEUE_NAME);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to create Destination: " , e);
}
return destination;
}
private MessageProducer createMessageProducer(Session session, Destination destination){
MessageProducer producer = null;
try {
producer = session.createProducer(destination);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to create MessageProducer: " , e);
}
return producer;
}
public void write(String text){
Connection connection = createConnection();
Session session = createSession(connection);
Destination destination = createDestination(session);
MessageProducer producer = createMessageProducer(session, destination);
Message message = null;
try {
if(text != null){
message = session.createTextMessage(text);
} else{
message = session.createTextMessage(TEXT);
}
producer.send(message);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to write Message to Destination: " , e);
} finally {
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to close the connection: " , e);
}
}
}
}
}