Table of Contents

Introduction

This article describes how to implement Java a JS7 workflow with two shell jobs that communicate over a Message Queue Service (MQ). These jobs include a one job for publishing for publishing and one job for receiving and executing JobScheduler API commands JS7 - REST Web Service API calls.

The document explains which classes and methods have to be created and their respective purposes. This example was has been developed with the use of Apache MQ.

...

for use with https://activemq.apache.org/.

Purpose of the Example

The example covers the specifics required to achieve the use case described below. It neither covers It neither covers the complete job JS7 REST Web Service API nor the complete possibilities of JMS.

For the coverage of the JobScheduler job JS7 REST Web Service API refer to the XML part the Technical Documentation of the REST Web Service API Documentation article.

For the coverage of the JMS API refer to the Oracle Java JMS Documentation.

...

Use Case

The use case consists of includes the following steps:

Run a Java Job in the JobScheduler which sends an XML fragment to a message queue Running a shell job which executes a Java class on an Agent. This sends a JSON body to a Message Queue (MQ)
Run a Java Job in the JobScheduler which receives an XML fragment from an MQ
Execute an XML command in the JobScheduler

What does this example explain?

The steps described in this article are as follows:

How to implement a Java job using the job API
How to implement a Producer Job
How to implement a Consumer Job
How to execute an XML fragment using the job API
How to deploy the jobs to the JobScheduler
How to configure the JobScheduler Jobs with JOE using the developed classes

Download

The classes described in this example can be downloaded here.

The job configuration (made with JOE) can be downloaded here.

Prerequisites

The following dependency is needed to write a Java job for the JobScheduler.

engine-job-api.jar
- The library is hosted on Maven Central. Please download the jar and add it to the classpath of the Java project.
- If the Java project already is a Maven project, simply add the following dependency to the project configuration.

Code Block

language	xml
title	Maven Dependency for engine-job-api
collapse	true

<dependency>
    <groupId>com.sos-berlin.jobscheduler.engine</groupId>
    <artifactId>engine-job-api</artifactId>
    <version>1.10.3</version>
</dependency>

- Make sure to use the correct version suitable for the JobScheduler version in use.

Running a shell job which executes a Java class on an Agent. This receives a JSON body from a Message Queue and executes a JS7 REST Web Service API call with the given JSON body.

What the example explains

This article shows the workflow configuration with the sample implementation as explained in the JS7 - Example - How to send and receive messages with a JMS Message Queue Service article.

Download

Download the library used in this example: js7-jms-example-jar.zip.
The archive contains:

the js7-jms-example.jar used in this example,
the js7-jms-example.config configuration file containing credentials for the connection and information about the workflow.

Download the workflow configuration (upload .json): js7-jms-example-workflow.zip.

Prerequisites

The activemq-all-5.15For this example the activemq-all-5.13.0.jar library is used for this example.

Either download the .jar file from the Active MQ web site and add it to the classpathclass path
or, in the case of a Maven project, add the following dependency to the project configuration.

Code Block

language	xml
title	Maven dependency for activemq-all
collapse	true

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.1315.0</version>
</dependency>

The Basic Structure of an API Job Java Class

A Java Job using the JobScheduler API has to extend the Job_impl class.

...

Adjust the js7-jms-example.config configuration file to your own environment.

Code Block

title	JMS configuration file: js7-jms-example.config
linenumbers	true
language	java
title	Base Structure of an API Job
collapse	true

public# classMessage CLASSNAMEQueue extendsService JobSettings
jms_implurl {

    @Override
    public boolean spooler_process() throws Exception {
		// This is the place where the work is done!
        return spooler_job.order_queue() != null;
    }
}

The Return Value of `spooler_process()`

If the Java job is an order job, it has to return true for the order to be able to continue with the next task if the job ends without an error.

If the Java job is a standalone job it has to return false for the JobScheduler to be able to recognize that the task has finished.

This can be achieved by using the return value spooler_job.order_queue() != null. It determines if an order is in the queue of the job (true) or not (false).

Logging

Because the Java Jobs run API methods, we can directly use the logging feature of the JobScheduler. As shown in the detailed method examples below, logging is done using the spooler_log method.

Make sure to throw the caught exception in order to hand it over to the spooler_process() method. This is needed for a job running in a job chain to determine that it has to fail if an error occurs.

Initialization of the MQ Connection

Both Java jobs need methods to initialize a connection to an MQ Server. This section of the document shows how to implement them. The implementation is based on the JMS implementation used and shipped by ActiveMQ.

To keep this example simple these methods are put in both Java job classes. To prevent duplicates in a real project´s source code, it is better to put these methods in a single class and extend the class in the two job classes.

The `createConnection()` method

This method instantiates a ConnectionFactory object with an ActiveMQConnectionFactory object and creates a Connection object through the factory method createConnection().

The ActiveMQConnectionFactory object has to be instantiated with the URL of the MQ server.

= tcp://localhost:61616
jms_queue_name      = test
jms_time_to_live    = 5000 

# JS7 JOC Cockpit Settings
joc_api_url         = http://localhost:4444/joc/api/
username            = root
password            = root

# /orders/add API Settings
controller_id       = js7
workflow_path       = /myWorkflowToStart

Workflow

The workflow in this example includes two jobs which both call the main class of the JmsExample.java class with either a produce or a consume argument.

the produce argument specifies the use of a SOSProducer to publish a message (JSON body) to a Message Queue Service.
the consume argument specifies the use of a SOSConsumer to read a message (JSON body) from a Message Queue Service andto:
login to the JS7 REST Web Service API,
send the API call /orders/add with the body received from the Message Queue Service and finally to
logout.

The Java executable has to be included in the PATH environment variable or the Java call has to be adjusted to include the path to the Java executable.

Note
Make sure that the `js7-jms-example.config` configuration file from the archive is stored in the same path as the `js7-jms-example.jar` library used in the workflow.

Code Block

language	js
title	Workflow Configuration
linenumbers	true
collapse	true

{
  "version": "1.1.0",
  "timeZone": "Europe/Berlin",
  "instructions": [
    {
      "TYPE": "Execute.Named",
      "jobName": "producer_job",
      "label": "job_produce"
    },
    {
      "TYPE": "Execute.Named",
      "jobName": "consumer_job",
      "label": "job_consume"
    }
  ],
  "jobs": {
    "producer_job":

Code Block

language	java
title	createConnection(String uri)
linenumbers	true
collapse	true

public Connection createConnection(String uri){
    ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
    Connection jmsConnection = null;
        try {
      "agentName": "standaloneAgent",
     jmsConnection = factory.createConnection();"executable": {
        } catch (JMSException e) {
"TYPE": "ShellScriptExecutable",
        "script": "java -jar C:/tmp/js7-jms-example.jar produce",
   spooler_log.error("JMSException occurred while trying to connect"v1Compatible": ");false
      },
      throw e;"skipIfNoAdmissionForOrderDay": false,
        }"parallelism": 1,
      "graceTimeout":  return jmsConnection;
}

The `createSession(Connection connection)` method

The method is called with an already instantiated Connection object and instantiates a Session object through the Connection object´s createSession(boolean transacted, int acknowledgeMode) method.

The acknowledgeMode has the following options among others:

AUTO_ACKNOWLEDGE
CLIENT_ACKNOWLEDGE

In summary the AUTO_ACKNOWLEDGE mode results in the message be dequeued when one consumer has read the message from the MQ server, whereas the CLIENT_ACKNOWLEDGE puts the responsibility on the client.

When the CLIENT_ACKNOWLEDGE mode is set, the message will stay present for all consumers to read until a consumer acknowledges the message. Only then the message will be dequeued and not be available for further consumers.

Code Block

language	java
title	createSession(Connection connection)
linenumbers	true
collapse	true

private Session createSession(Connection connection) throws JMSException{
    Session session = null;
    try {15,
      "failOnErrWritten": false,
      "warnOnErrWritten": false,
      "title": "produce"
    },
    "consumer_job": {
      "agentName": "standaloneAgent",
      "executable": {
        "TYPE": "ShellScriptExecutable",
        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
"script": "java -jar C:/tmp/js7-jms-example.jar consume",
     } catch (JMSException e) {
   "v1Compatible": false
      },
      spooler_log.error("JMSException occurred while trying to create Session: ");"skipIfNoAdmissionForOrderDay": false,
      "parallelism": 1,
      "graceTimeout": 15,
      "failOnErrWritten":  throw e;
false,
      "warnOnErrWritten": false,
       }"title": "consumer"
    }
 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

language	java
title	createDestination(Session session, String queueName)
linenumbers	true
collapse	true

private Destination createDestination(Session session, String queueName) throws JMSException{
    Destination destination = null;
    try {
        destination = session.createQueue(queueName);
    } catch (JMSException e) {
        spooler_log.error("JMSException occurred while trying to create Destination: ");
        throw e;
    }
    return destination;
}

The `createConnectionUrl (String protocol, String hostName, String port)` method

This is a helper method which creates a connection URI based on the job parameters. It simply uses a StringBuilder to create a string in the format [protocol]://[hostname]:[port].

Code Block

language	java
title	createConnectionUrl (String protocol, String hostName, String port)
linenumbers	true
collapse	true

private String createConnectionUrl (String protocol, String hostName, String port){
    StringBuilder strb = new StringBuilder();
    strb.append(protocol).append("://").append(hostName).append(":").append(port);
    return strb.toString();
}

Implementation of the Producer Job

There are three more methods needed for the producer Java job to run:

createMessageProducer(Session session, Destination destination)
write(String text, String connectionUrl, String queueName)
execute()

The `createMessageProducer(Session session, Destination destination)` method

This method is called with an already active Session object as well as an instantiated Destination object. It instantiates a MessageProducer object with the given session and destination.

Code Block

language	java
title	createMessageProducer(Session session, Destination destination)
linenumbers	true
collapse	true

public MessageProducer createMessageProducer(Session session, Destination destination) throws JMSException{
    MessageProducer producer = null;
    try {
        producer = session.createProducer(destination);
    } catch (JMSException e) {
        spooler_log.error("JMSException occurred while trying to create MessageProducer: ");
        throw e;
    }
    return producer;
}

The `write(String text, String connectionUrl, String queueName)` method

The method is called with the message to be sent to the server, the connection URL to the server and the queue name to be used for publishing. The message is a String object. The method instantiates a Message object with the text to send. This method makes use of the methods described above. It instantiates all objects needed for the connection and sends a Message object. Also the methods throws an error if one occurs. Although this method may throw an error, it still calls the MessageProducer´s send(Message message) method in a try..catch..finally block. That is because the finally block is used to close the connection in the end.

Code Block

language	java
title	write(String text, String connectionUrl, String queueName)
linenumbers	true
collapse	true

public void write(String text, String connectionUrl, String queueName) throws JMSException{
    Connection connection = createConnection(connectionUrl);
    Session session = createSession(connection);
    Destination destination = createDestination(session, queueName);
    MessageProducer producer = createMessageProducer(session, destination);
    Message message = null;
    try {
        message = session.createTextMessage(text);
        producer.send(message);
    } catch (JMSException e) {
        spooler_log.error("JMSException occurred in ProducerJob while trying to write Message to Destination: ");
        throw e;
    } finally {
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException e) {
                spooler_log.error("JMSException occurred in ProducerJob while trying to close the connection: ");
                throw e;
            }
        }
    }
}

The `execute()` method

This is the core method of the Producer Java job. It reads the parameters from the task and order, does some logging at debug level, calls the createConnectionUrl(..) method and subsequently calls the write(..) method to send the message to the MQ server.

Code Block

language	java
title	execute()
linenumbers	true
collapse	true

private void execute() throws Exception {
    Variable_set params = spooler_task.params();
    params.merge(spooler_task.order().params());
    spooler_log.debug9(params.xml());
    String protocol = spooler_task.params().value("MQ_Protocol");
    spooler_log.debug9("Received protocol: " + protocol);
    String messageHost = spooler_task.params().value("MQ_Host");
    spooler_log.debug9("Received MQ Host: " + messageHost);
    String messagePort = spooler_task.params().value("MQ_Port");
    spooler_log.debug9("Received MQ port: " + messagePort);
    String queueName = spooler_task.params().value("MQ_QueueName");
    spooler_log.debug9("Received Queue name: " + queueName);
    String message = spooler_task.params().value("message");
    spooler_log.debug9("Received message: " + message);
    if(protocol == null || (protocol != null && protocol.isEmpty())){
        protocol = DEFAULT_PROTOCOL;
    }
    if(queueName == null || (queueName != null && queueName.isEmpty())){
        queueName = DEFAULT_QUEUE_NAME;
    }
    String connectionUrl = createConnectionUrl(protocol, messageHost, messagePort);
    if(message != null && !message.isEmpty()){
        write(message, connectionUrl, queueName);
    } else {
        spooler_log.error("Message is empty, nothing to send to message server");
    }
}

Putting the Bricks Together

The last thing to do is to call the execute() method in spooler_process() as shown in the complete code example below.

Code Block

language	java
title	Complete Source of the MessageProducerExampleJob
linenumbers	true
collapse	true

package com.sos.jitl.messaging;

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 sos.spooler.Job_impl;
import sos.spooler.Variable_set;

public class MessageProducerExampleJob extends Job_impl {

    private static final String DEFAULT_QUEUE_NAME = "JobChainQueue";
    private static final String DEFAULT_PROTOCOL = "tcp";

    @Override
    public boolean spooler_process() throws Exception {
        try {
            execute();
        } catch (Exception e) {
            spooler_log.error("Error occured in spooler_process() of MessageProducerExampleJob: ");
            throw e;
        }
        return  spooler_job.order_queue() != null;
    }

    private void execute() throws Exception {
        Variable_set params = spooler_task.params();
        params.merge(spooler_task.order().params());
        spooler_log.debug9(params.xml());
        String protocol = spooler_task.params().value("MQ_Protocol");
        spooler_log.debug9("Received protocol: " + protocol);
        String messageHost = spooler_task.params().value("MQ_Host");
        spooler_log.debug9("Received MQ Host: " + messageHost);
        String messagePort = spooler_task.params().value("MQ_Port");
        spooler_log.debug9("Received MQ port: " + messagePort);
        String queueName = spooler_task.params().value("MQ_QueueName");
        spooler_log.debug9("Received Queue name: " + queueName);
        String message = spooler_task.params().value("message");
        spooler_log.debug9("Received message: " + message);
        if(protocol == null || (protocol != null && protocol.isEmpty())){
            protocol = DEFAULT_PROTOCOL;
        }
        if(queueName == null || (queueName != null && queueName.isEmpty())){
            queueName = DEFAULT_QUEUE_NAME;
        }
        String connectionUrl = createConnectionUrl(protocol, messageHost, messagePort);
        if(message != null && !message.isEmpty()){
            write(message, connectionUrl, queueName);
        } else {
            spooler_log.error("Message is empty, nothing to send to message server");
        }
    }

    private String createConnectionUrl (String protocol, String hostName, String port){
        StringBuilder strb = new StringBuilder();
        strb.append(protocol).append("://").append(hostName).append(":").append(port);
        return strb.toString();
    }

    private Connection createConnection(String uri) throws JMSException{
        ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
        Connection jmsConnection = null;
        try {
            jmsConnection = factory.createConnection();
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to connect: ");
            throw e;
        }
        return jmsConnection;
    }
    
    private Session createSession(Connection connection) throws JMSException{
        Session session = null;
        try {
            session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to create Session: ");
            throw e;
        }
        return session;
    }
    
    private Destination createDestination(Session session, String queueName) throws JMSException{
        Destination destination = null;
        try {
            destination = session.createQueue(queueName);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to create Destination: ");
            throw e;
        }
        return destination;
    }
    
    public MessageProducer createMessageProducer(Session session, Destination destination) throws JMSException{
        MessageProducer producer = null;
        try {
            producer = session.createProducer(destination);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to create MessageProducer: ");
            throw e;
        }
        return producer;
    }
    
    public void write(String text, String connectionUrl, String queueName) throws JMSException{
        Connection connection = createConnection(connectionUrl);
        Session session = createSession(connection);
        Destination destination = createDestination(session, queueName);
        MessageProducer producer = createMessageProducer(session, destination);
        Message message = null;
        try {
            message = session.createTextMessage(text);
            producer.send(message);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred in ProducerJob while trying to write Message to Destination: ");
            throw e;
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    spooler_log.error("JMSException occurred in ProducerJob while trying to close the connection: ");
                    throw e;
                }
            }
        }
    }
    
}

Implementation of the Consumer Job

As seen above in the Producer Job example, there are also additional methods needed for the consumer Java job to run.

createMessageConsumer(Session session, Destination destination)
read(Connection connection, String queueName, Boolean closeMessage)
executeXml(String message)
executeXmlForAllTargets(String message)
execute()

The `createMessageConsumer(Session session, Destination destination)` method

This method is called with an already active Session object as well as an instantiated Destination object. It instantiates a MessageConsumer object with the given session and destination.

Code Block

language	java
title	createMessageConsumer(Session session, Destination destination)
linenumbers	true
collapse	true

private MessageConsumer createMessageConsumer(Session session, Destination destination) throws JMSException {
    MessageConsumer consumer = null;
    try {
        consumer = session.createConsumer(destination);
    } catch (JMSException e) {
        spooler_log.error("JMSException occurred while trying to create MessageConsumer: ");
        throw e;
    }
    return consumer;
}

The `read(Connection connection, String queueName, Boolean closeMessage)` method

The method is called with an already instantiated Connection object, the name of the queue to read from and a flag to determine if this is the last consumer. If it is the last consumer for the messages in the specified queue then the received message will be acknowledged. This method makes use of the methods described above. It instantiates all objects needed for the connection and reads a Message object from the given queue. It extracts the value from the Message object as a string representation via the Message object´s getText() method. We have to make sure that connection.start() is called before the MessageConsumer object is instantiated.

Code Block

language	java
title	read(Connection connection, String queueName, Boolean closeMessage)
linenumbers	true
collapse	true

private String read(Connection connection, String queueName, Boolean closeMessage) throws JMSException {
    String messageText = null;
    try {
        Session session = createSession(connection);
        Destination destination = createDestination(session, queueName);
        connection.start();
        MessageConsumer consumer = createMessageConsumer(session, destination);
        Message receivedMessage = null;
        while (true) {
            receivedMessage = consumer.receive(1);
            if (receivedMessage != null) {
                if (receivedMessage instanceof TextMessage) {
                    TextMessage message = (TextMessage) receivedMessage;
                    messageText = message.getText();
                    break;
                } else {
                    break;
                }
            }
        }
        if(closeMessage){
            receivedMessage.acknowledge();
        }
    } catch (JMSException e) {
        spooler_log.error("JMSException occurred while trying to read from Destination: ");
        throw e;
    } finally {
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException e) {
                spooler_log.error("JMSException occurred while trying to close the connection: ");
            }
        }
    }
    return messageText;
}

The `executeXml(String message)` method

This method gets the received message and sends it through the spooler.executeXml(String text) API method with additional debug information.

Code Block

language	java
title	executeXml(String message)
linenumbers	true
collapse	true

private void executeXml(String message) {
    spooler_log.debug9("execute XML started");
    String answer = spooler.execute_xml(message);
    spooler_log.debug9("Return value of executeXML: " + answer);
    spooler_log.debug9("execute XML finished");
}

The `executeXmlForAllTargets(String message)` method

This method calls the executeXml(String message) method mentioned above for each given target job chain. It replaces the job_chain argument of the (XML) message with the given job chain names from the job parameter.

Code Block

language	java
title	executeXmlForAllTargets(String message)
linenumbers	true
collapse	true

private void executeXmlForAllTargets(String message) {
    if (targetJobChains.contains(DELIMITER) && message.contains("add_order")) {
        String [] jobChainNames = targetJobChains.split("[" + DELIMITER + "]");
        for (String name : jobChainNames){
            spooler_log.debug9("add_order XML will be adjusted for JobChain: " + name);
            executeXml(message.replaceFirst("job_chain='[^']*'", "job_chain='" + name + "'"));
        }            
    }else{
        executeXml(message);
    }
}

The `execute()` method

This is the core method of the Producer Java job. It reads the parameters from the task and order, does some logging at debug level, calls the createConnectionUrl(..) method and subsequently calls the read(..) method. The received message as well as the name(s) of the target job chain(s) are stored globally in the class for further processing.

Code Block

language	java
title	execute()
linenumbers	true
collapse	true

private void execute() throws Exception {
    Variable_set params = spooler_task.params();
    params.merge(spooler_task.order().params());
    spooler_log.debug9(params.xml());
    String protocol = params.value("MQ_Protocol");
    spooler_log.debug9("Received protocol: " + protocol);
    if(protocol == null || (protocol != null && protocol.isEmpty())){
        protocol = DEFAULT_PROTOCOL;
    }
    String messageHost = params.value("MQ_Host");
    spooler_log.debug9("Received MQ Host: " + messageHost);
    String messagePort = params.value("MQ_Port");
    spooler_log.debug9("Received MQ port: " + messagePort);
    String queueName = params.value("MQ_QueueName");
    spooler_log.debug9("Received Queue name: " + queueName);
    Boolean lastConsumer = Boolean.valueOf(params.value("lastConsumer"));
    spooler_log.debug9("Received lastConsumer: " + lastConsumer.toString());
    if(queueName == null || (queueName != null && queueName.isEmpty())){
        queueName = DEFAULT_QUEUE_NAME;
    }
    targetJobChains = params.value("targetJobChainName");
    spooler_log.debug9("Received targetJobChains: " + targetJobChains);
    String connectionUrl = createConnectionUrl(protocol, messageHost, messagePort);
    Connection jmsConnection = createConnection(connectionUrl);
    messageXml = read(jmsConnection, queueName, lastConsumer);
    spooler_log.debug9("Received message: " + messageXml);
}

Putting the Bricks Together

The last thing to do is for the Consumer Job is to call the execute() method as well as the executeXmlForAllTargets(String message) method in spooler_process(), as shown in the complete code example below.

Code Block

language	java
title	Complete Source of the MessageConsumerExampleJob
linenumbers	true
collapse	true

package com.sos.jitl.messaging;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import sos.spooler.Job_impl;
import sos.spooler.Variable_set;

public class MessageConsumerExampleJob extends Job_impl {

    private static final String DEFAULT_QUEUE_NAME = "JobChainQueue";
    private static final String DEFAULT_PROTOCOL = "tcp";
    private static final String DELIMITER = ",";
    private String messageXml; 
    private String targetJobChains;

    @Override
    public boolean spooler_process() throws Exception {
        try {
            execute();
            executeXmlForAllTargets(messageXml);
        } catch (Exception e) {
            spooler_log.error("Error occured in spooler_process() of MessageConsumerTestJob: ");
            throw e;
        }
        return spooler_job.order_queue() != null;
    }

    private void execute() throws Exception {
        Variable_set params = spooler_task.params();
        params.merge(spooler_task.order().params());
        spooler_log.debug9(params.xml());
        String protocol = params.value("MQ_Protocol");
        spooler_log.debug9("Received protocol: " + protocol);
        if(protocol == null || (protocol != null && protocol.isEmpty())){
            protocol = DEFAULT_PROTOCOL;
        }
        String messageHost = params.value("MQ_Host");
        spooler_log.debug9("Received MQ Host: " + messageHost);
        String messagePort = params.value("MQ_Port");
        spooler_log.debug9("Received MQ port: " + messagePort);
        String queueName = params.value("MQ_QueueName");
        spooler_log.debug9("Received Queue name: " + queueName);
        Boolean lastConsumer = Boolean.valueOf(params.value("lastConsumer"));
        spooler_log.debug9("Received lastConsumer: " + lastConsumer.toString());
        if(queueName == null || (queueName != null && queueName.isEmpty())){
            queueName = DEFAULT_QUEUE_NAME;
        }
        targetJobChains = params.value("targetJobChainName");
        spooler_log.debug9("Received targetJobChains: " + targetJobChains);
        String connectionUrl = createConnectionUrl(protocol, messageHost, messagePort);
        Connection jmsConnection = createConnection(connectionUrl);
        messageXml = read(jmsConnection, queueName, lastConsumer);
        spooler_log.debug9("Received message: " + messageXml);
    }
    
    private String createConnectionUrl (String protocol, String hostName, String port){
        StringBuilder strb = new StringBuilder();
        strb.append(protocol).append("://").append(hostName).append(":").append(port);
        return strb.toString();
    }

    private Connection createConnection(String uri) throws JMSException{
        ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
        Connection jmsConnection = null;
        try {
            jmsConnection = factory.createConnection();
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to connect: ");
            throw e;
        }
        return jmsConnection;
    }
    
    private Session createSession(Connection connection) throws JMSException{
        Session session = null;
        try {
            session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to create Session: ");
            throw e;
        }
        return session;
    }
    
    private Destination createDestination(Session session, String queueName) throws JMSException{
        Destination destination = null;
        try {
            destination = session.createQueue(queueName);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to create Destination: ");
            throw e;
        }
        return destination;
    }
    
    private MessageConsumer createMessageConsumer(Session session, Destination destination) throws JMSException {
        MessageConsumer consumer = null;
        try {
            consumer = session.createConsumer(destination);
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to create MessageConsumer: ");
            throw e;
        }
        return consumer;
    }

    private String read(Connection connection, String queueName, Boolean closeMessage) throws JMSException {
        String messageText = null;
        try {
            Session session = createSession(connection);
            Destination destination = createDestination(session, queueName);
            connection.start();
            MessageConsumer consumer = createMessageConsumer(session, destination);
            Message receivedMessage = null;
            while (true) {
                receivedMessage = consumer.receive(1);
                if (receivedMessage != null) {
                    if (receivedMessage instanceof TextMessage) {
                        TextMessage message = (TextMessage) receivedMessage;
                        messageText = message.getText();
                        break;
                    } else {
                        break;
                    }
                }
            }
            if(closeMessage){
                receivedMessage.acknowledge();
            }
        } catch (JMSException e) {
            spooler_log.error("JMSException occurred while trying to read from Destination: ");
            throw e;
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    spooler_log.error("JMSException occurred while trying to close the connection: ");
                }
            }
         }
        return messageText;
    }

    private void executeXml(String message) {
        spooler_log.debug9("execute XML started");
        String answer = spooler.execute_xml(message);
        spooler_log.debug9("Return value of executeXML: " + answer);
        spooler_log.debug9("order sent");
    }
    
    private void executeXmlForAllTargets(String message) {
        if (targetJobChains.contains(DELIMITER) && message.contains("add_order")) {
            String [] jobChainNames = targetJobChains.split("[" + DELIMITER + "]");
            for (String name : jobChainNames){
                spooler_log.debug9("add_order XML will be adjusted for JobChain: " + name);
                executeXml(message.replaceFirst("job_chain='[^']*'", "job_chain='" + name + "'"));
            }            
        }else{
            executeXml(message);
        }
    }
}

Deploy the Java Classes to the JobScheduler

To be able to run the new Java jobs simply deploy/copy the created jar library to the user_lib folder in your $SCHEDULER_HOME.

Configuring the Jobs in JOE

This section shows the creation of an example job chain to run both jobs.

Create the Producer Job in JOE

Create a new Job and write the class name including the package structure in the Classname field. Let´s call the job XML_Producer after the example's use case.

Image Removed

Configure the parameters for the job as shown below.

Image Removed

Create the Consumer Job in JOE

Create a new Job and write the class name including the package structure in the Classname field. Let´s call the job XML_Consumer after the example's use case.

Image Removed

Configure the parameters for the job as shown below.

Image Removed

The targetJobChainName parameter is only needed if you want to change the configuration as shown in the message parameter of the producer job.

Create the Job Chain with JOE

This job chain configuration has been written to show the processing of the job - therefore both jobs are put in the same job chain. In a production environment it is more likely that both jobs would reside in different job chains, that there is more than one consumer, etc...

Create a job chain which looks like the example below.

...

}
}

Space shortcuts

Page tree

Page History

Versions Compared

Old Version 2

New Version Current

Key

Introduction

Purpose of the Example

Use Case

What does this example explain?

Download

Prerequisites

What the example explains

Download

Prerequisites

The Basic Structure of an API Job Java Class

The Return Value of spooler_process()

Logging

Initialization of the MQ Connection

The createConnection() method

Workflow

The createSession(Connection connection) method

The createDestination(Session session, String queueName) method

The createConnectionUrl (String protocol, String hostName, String port) method

Implementation of the Producer Job

The createMessageProducer(Session session, Destination destination) method

The write(String text, String connectionUrl, String queueName) method

The execute() method

Putting the Bricks Together

Implementation of the Consumer Job

The createMessageConsumer(Session session, Destination destination) method

The read(Connection connection, String queueName, Boolean closeMessage) method

The executeXml(String message) method

The executeXmlForAllTargets(String message) method

The execute() method

Putting the Bricks Together

Deploy the Java Classes to the JobScheduler

Configuring the Jobs in JOE

Create the Producer Job in JOE

Create the Consumer Job in JOE

Create the Job Chain with JOE

The Return Value of `spooler_process()`

The `createConnection()` method

The `createSession(Connection connection)` method

The `createDestination(Session session, String queueName)` method

The `createConnectionUrl (String protocol, String hostName, String port)` method

The `createMessageProducer(Session session, Destination destination)` method

The `write(String text, String connectionUrl, String queueName)` method

The `execute()` method

The `createMessageConsumer(Session session, Destination destination)` method

The `read(Connection connection, String queueName, Boolean closeMessage)` method

The `executeXml(String message)` method

The `executeXmlForAllTargets(String message)` method

The `execute()` method