...
To make the implementation more readable as well as reusable, the steps are separated in different methods.
...
createConnection(String url)
...
This method instantiates a ConnectionFactory object with an ActiveMQConnectionFactory object and creates a Connection object through the factory method createConnection().
...
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
public 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;
} |
...
createSession(Connection connection)
...
The method is called with an already instantiated Connection object and initiates a Session object through the 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;
} |
...
createDestination(Session session, String queueName)
...
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;
} |
...
createMessageProducer(Session session, Destination destination)
...
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;
} |
...
write(String text, MessageProducer producer)
)...
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);
}
} |
...
close()
...
This methods makes sure that the used connection will be closed after a message is sent.
...
This part of the document will show examples for the last 5 steps as the first four steps are simmilar as to the examples shown above for the instantiation of the MessageProducer.
The Methods
...
createMessageConsumer(Session session, Destination destination)
...
This method is called with an already active Session object as well as instantiated Destination object. It instantiates a MessageConsumer object with the given session and destination.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
private MessageConsumer createMessageConsumer(Session session, Destination destination) {
MessageConsumer consumer = null;
try {
consumer = session.createConsumer(destination);
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to create MessageConsumer: ", e);
}
return consumer;
} |
...
read(MessageConsumer consumer)
...
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
private String read(MessageConsumer consumer) {
TextMessage message = null;
String textMessage = null;
try {
while (true) {
Message receivedMessage = consumer.receive(1);
if (receivedMessage != null) {
if (receivedMessage instanceof TextMessage) {
message = (TextMessage) receivedMessage;
textMessage = message.getText();
LOGGER.info("Reading message: " + textMessage);
break;
} else {
break;
}
}
}
} catch (JMSException e) {
LOGGER.error("JMSException occurred while trying to read from Destination: ", e);
}
return textMessage;
} |
| Note |
|---|
Don´t forget to clean up (call |
...
connect(String host, int port)
)...
This method instantiates a TCP socket connection to a JobScheduler instance with the given host:port.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
public void connect(String host, int port) throws Exception {
if (host == null || host.length() == 0) {
throw (new Exception("hostname missing."));
}
if (port == 0){
throw (new Exception("port missing."));
}
if ("udp".equalsIgnoreCase(PROTOCOL)) {
udpSocket = new DatagramSocket();
udpSocket.connect(InetAddress.getByName(HOST), PORT);
} else {
socket = new Socket(host, port);
in = new DataInputStream(socket.getInputStream());
out = new PrintWriter(socket.getOutputStream(), true);
}
} |
...
disconnect()
...
This method is used to close the socket opened with the above connect() method.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
public void disconnect() throws Exception {
if (socket != null) {
socket.close();
}
if (in != null) {
in.close();
}
if (out != null) {
out.close();
}
} |
...
sendRequest(String command)
...
This method sends a request with the given command to the connected JobScheduler instance. The given command in the example is the add_order XML snippet shown in the SOSProducer example.
| Code Block | ||||||
|---|---|---|---|---|---|---|
| ||||||
public void sendRequest(String command) throws Exception {
if ("udp".equalsIgnoreCase(PROTOCOL)) {
if (command.indexOf("<?xml") == -1) {
command = "<?xml version=\"1.0\" encoding=\"iso-8859-1\"?>" + command + "\r\n";
}
byte[] commandBytes = command.getBytes();
udpSocket.send(new DatagramPacket(commandBytes, commandBytes.length, InetAddress.getByName(HOST), PORT));
} else {
if (command.indexOf("<?xml") == 0) {
out.print(command + "\r\n");
} else {
out.print("<?xml version=\"1.0\" encoding=\"iso-8859-1\"?>" + command + "\r\n");
}
out.flush();
}
} |
...
getResponse()
...
THis message receives the response of the TCP connection described above.
| Code Block | ||||||
|---|---|---|---|---|---|---|
| ||||||
public String getResponse() throws IOException, RuntimeException {
int sec = 0;
byte[] buffer = {};
while (in.available() == 0) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
LOGGER.error("", e);
}
if (sec++ == TIMEOUT) {
throw new RuntimeException("timeout reached");
}
}
buffer = new byte[in.available()];
int bytesRead = in.read(buffer);
return new String(buffer, "ISO-8859-1");
} |
...
receiveFromQueue()
...
The receiveFromQueue() method uses the methods described above to connect to a Host running a JobScheduler instance, read from a message queue, send the received message to the JobScheduler instance and close the session to the JobScheduler instance.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
public String receiveFromQueue() {
String message = null;
try {
connect();
message = read();
sendRequest(message);
disconnect();
} catch (Exception e) {
LOGGER.error("Error occured while publishing to the JobScheduler instance host:" + HOST + ", port:" + PORT, e);
}
return message;
} |
The SOSConsumer class
The code example below slightly differs from the examples above. In the class below the methode read() already uses the other method for instantiation, therefore it needs less parameters and it closes the connection itself.
...