...
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 ProducerMessageProducer
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.
...
This method instantiates a ConnectionFactory object with an ActiveMQConnectionFactory object and creates a connection Connection object through the factory method createConnection().
The ActiveMQConnectionFactory has object has to be instantiated with the URL of the MQ server.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
privatepublic 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 method is called with an already instantiated connection Connection object and starts initiates a session Session object through the connection objects Connection objects createSession(boolean transacted, int acknowledgeMode) method.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
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, String queueName) method
This method creates a Destination object. It is called with an active Session object and the name of the queue to write to. The Destination object is initiated through the createQueue(String name) method of the Session object.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
public Destination createDestination(Session session, String queueName){
Destination destination = null;
try {
destination = session.createQueue(queueName);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to create Destination: " , e);
}
return destination;
} |
The createMessageProducer(Session session, Destination destination) method
...
This method is called with an already active Session object as well as instantiated Destination object. It instantiates a MessageProducer object with the given session and destination.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
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;
} |
The write(String text, MessageProducer producer) method
)The method is called with the message to send to the server and the MessageProducer object to use for publishing. The message is a String object. The method instantiates a Message object with the text to send.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
public void write(String text, MessageProducer producer){
Message message = null;
try {
if(text != null){
message = session.createTextMessage(text);
} else{
message = session.createTextMessage(DEFAULT_TEXT);
}
producer.send(message);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to write Message to Destination: " , e);
}
} |
The close() method
This methods makes sure that the used connection will be closed after a message is sent.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
private void close(Connection connection){
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to close the connection: " , e);
}
}
} |
The SOSProducer class
The code example below slightly differs from the examples above. In the class below the methode write(String text) already uses the other method for instantiation, therfore it needs less parameters and it closes the connection itself.
The text in the example below consists of a JobScheduler add_order XML, which can later be used to send to a JobScheduler.TBD
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
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 DEFAULT_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);
}
privatepublic 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, String queueName){
Destination destination = null;
try {
destination = session.createQueue(QUEUE_NAMEqueueName);
} 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(DEFAULT_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);
}
}
}
}
} |
...
} |
The MessageConsumer
This example describes how to build a connection to a MQ server and receive a message from a queue on this server. This example uses Apache Active MQ as the message queue server.
- Create a connection.
- Create a session.
- Create a destination.
- Create a consumer.
- Receive a message with the consumer.
- Close the (MQ) connection.
- Open a TCP connection to a JobScheduler instance.
- Send the Message to the JobScheduler
- Close the (TCP) connection.