7.6.1. MOM Messaging¶

7.6.1.1.1. Compatible with various MOMs¶

Since MOM messaging various MOMs, the MessagingProvider interface is provided. MQ connection that depends on MOM and message send/receive are performed by the class that implements this interface. Therefore, this function can be used in various MOMs by creating a class that implements the MessagingProvider interface.

MOM messaging is compatible with JMS and offers JmsMessagingProvider. For details, refer to the linked Javadoc.

It also supports IBM MQ, which is widely used as a MOM. For details, see IBM MQ Adapter.

7.6.1.3.1. Configure settings to use MOM messaging¶

In MOM messaging, add the following classes to the component definition.

• Implementation class MessagingProvider (MQ connection, MQ send to/receive from)
• Messaging Context Management Handler (MQ connection management)

A configuration example is shown below.

<!-- Implementation class of MessagingProvider -->
<component name="messagingProvider"
           class="nablarch.fw.messaging.provider.JmsMessagingProvider">
  <!-- See Javadoc for configuration items-->
</component>

<!-- Messaging context management handler -->
<component name="messagingContextHandler"
           class="nablarch.fw.messaging.handler.MessagingContextHandler">
  <property name="messagingProvider" ref="messagingProvider" />
</component>

In the case of messages are received, it is necessary to set the data reader. Add the following class to the component definition.

• MessageReader (reads messages from MQ)
• FwHeaderReader (reads the framework control header from the message)

A configuration example is shown below.

Point

• Specify dataReader in the component name of the data reader.
• Specify MessageReader in the messageReader property of FwHeaderReader.

<!-- FwHeaderReader -->
<component name="dataReader"
           class="nablarch.fw.messaging.reader.FwHeaderReader">
  <!-- MessageReader -->
  <property name="messageReader">
    <component class = "nablarch.fw.messaging.reader.MessageReader">
      <!-- See Javadoc for configuration items-->
    </component>
  </property>
</component>

7.6.1.3.2. Send message with asynchronous response (Sending asynchronous message)¶

Send a message to an external system.

Contents of Common protocol header Common protocol header configured in send message

Basically, only the send address header needs to be set.

Message ID:	Configuration not required (numbered after sending)
Correlation message ID:
	Configuration not required
Send address:	Logical name of the send address
Reply to address:
	Configuration not required
Expiry interval:
	Any

Sending asynchronous message provides AsyncMessageSendAction as a common action to acquire the data to be sent from the table (called a temporary table) held by the send message data, and create and send messages. AsyncMessageSendAction is an action class that works with Nablarch Batch Application.

Only the following deliverables are required to be created using AsyncMessageSendAction and the message send process can be implemented very easily.

• Temporary table that holds the data of send message
• Format definition file that represents the message layout
• SQL file (defines three types of SQL statements)

• SELECT statement for acquiring data with not sent status
• UPDATE statement to update the status of the relevant data to processed when the message is sent successfully
• UPDATE statement for updating the status of the relevant data to send failure when the message send fails

• Form class for status update

Below is an implementation example based on Example application to send project information.

Implementation examples

Temporary table that holds the data of send message

Point

• The primary key is the column that stores the ID that uniquely identifies the message.
• The attribute information of the table defines the columns corresponding to each item of the message to be sent.
• Define common items (update user ID, update date, etc.) according to each project architecture.

INS_PROJECT_SEND_MESSAGE

Send message serial number (PK)	SEND_MESSAGE_SEQUENCE
Project name	PROJECT_NAME
Project type	PROJECT_TYPE
Project classification	PROJECT_CLASS
: (Omitted)
Status	STATUS
Update user ID	UPDATED_USER_ID
Update date and time	UPDATED_DATE

Format definition file

Point

• The file name is <Request ID of send message>_SEND.fmt.

ProjectInsertMessage_SEND.fmt

file-type:        "Fixed" # Fixed-length
text-encoding:    "MS932" # Character encoding of string type field
record-length:    2120    # Length of each record

[userData]
Item definition omitted

SQL file

Point

• The file name is <Request ID of send message>.sql.
• SQL_ID is as follows.

• SELECT_SEND_DATA: SELECT statement for acquiring data with not sent status
• UPDATE_NORMAL_END: UPDATE statement to update the status to processed
• UPDATE_ABNORMAL_END: UPDATE statement to update the status to send failure

ProjectInsertMessage.sql

SELECT_SEND_DATA =
SELECT
    Omitted
FROM
    INS_PROJECT_SEND_MESSAGE
WHERE
    STATUS = '0'
ORDER BY
    SEND_MESSAGE_SEQUENCE

UPDATE_NORMAL_END =
UPDATE
    INS_PROJECT_SEND_MESSAGE
SET
    STATUS = '1',
    UPDATED_USER_ID = :updatedUserId,
    UPDATED_DATE = :updatedDate
WHERE
    SEND_MESSAGE_SEQUENCE = :sendMessageSequence

UPDATE_ABNORMAL_END =
UPDATE
    INS_PROJECT_SEND_MESSAGE
SET
    STATUS = '9',
    UPDATED_USER_ID = :updatedUserId,
    UPDATED_DATE = :updatedDate
WHERE
    SEND_MESSAGE_SEQUENCE = :sendMessageSequence

Form class for status update

Point

• Since this form class is for status update only, all the attributes of the temporary table are not required to be retained as properties.

SendMessagingForm.java

public class SendMessagingForm {

    /** Send message serial number */
    private String sendMessageSequence;

    /** Update user ID */
    @UserId
    private String updatedUserId;

    /** Update date and time */
    @CurrentDateTime
    private java.sql.Timestamp updatedDate;

    // Constructor and accessor are omitted
}

Configuration of AsyncMessageSendAction

Point

• When AsyncMessageSendAction is used, configuration of queue name of send destination and storage directory of format definition file are required. Configured by adding AsyncMessageSendActionSettings to the component definition. For the configuration items, refer to the Javadoc link.

messaging-async-send-component-configuration.xml

<component name="asyncMessageSendActionSettings"
           class="nablarch.fw.messaging.action.AsyncMessageSendActionSettings">
  <property name="formatDir" value="format" />
  <property name="headerFormatName" value="header" />
  <property name="queueName" value="TEST.REQUEST" />
  <property name="sqlFilePackage" value="com.nablarch.example.sql" />
  <property name="formClassName"
            value="com.nablarch.example.form.SendMessagingForm" />
  <property name="headerItemList">
    <list>
      <value>sendMessageSequence</value>
    </list>
  </property>
</component>

Applying AsyncMessageSendAction

Point

• To make AsyncMessageSendAction work with Nablarch Batch Application, specify AsyncMessageSendAction in the component definition of Request Dispatch Handler.

messaging-async-send-component-configuration.xml

<component class="nablarch.fw.handler.RequestPathJavaPackageMapping">
  <property name="basePackage"
            value="com.nablarch.example.action.ExampleAsyncMessageSendAction" />
  <property name="immediate" value="false" />
</component>

7.6.1.3.3. Send message with synchronous response (Sending synchronous message)¶

Send a message to an external system and wait for the reply. Block until a response message is received or the wait timeout expires.

Since a response message is received unlike Send message with asynchronous response (Sending asynchronous message), it can be guaranteed to some extent that the process worked correctly at the destination. However, if a response is not received within the specified time due to some problem and a timeout occurs, it is necessary to perform error processing (for example, message retry or failure notification).

Contents of Common protocol header configured in send message

In addition to the send address header, it is necessary to set the reply to address header, which is the send address for the response.

Message ID:	Configuration not required (numbered after sending)
Correlation message ID:
	Configuration not required
Send address:	Logical name of the send address
Reply to address:
	Logical name of reply to address
Expiry interval:
	Any

Content of Common protocol header of response message created by the external system

After the send process is completed, the application waits until a message with the same correlation message ID as that of the send message is received at the reply to address. Therefore, the external system must configure the correlation message ID in the response message.

Message ID:	Configuration not required (numbered after sending)
Correlation message ID:
	Value of message ID header of send message
Send address:	Reply to address header of send message
Reply to address:
	Configuration not required
Expiry interval:
	Any

Sending synchronous message provides MessageSender as a utility class that wraps routine processing. Only the following deliverables are required to be created using MessageSender and the send synchronous response message can be created very easily.

• Format definition file used to send and receive
• Send/receive process with MessageSender

The following is an implementation example of sending project information by batch action from the data stored in the table based on Example application. Since the section to read data from the table is not related to sending of messages, the implementation example is omitted.

Implementation examples

Format definition file used to send and receive

Point

• The file name is as follows.

• To send: <Message request ID>_SEND.fmt
• To receive: <Message request ID>_RECEIVE.fmt

• The record type name is fixed to data.

ProjectInsertMessage_SEND.fmt

file-type:        "Fixed" # Fixed-length
text-encoding:    "MS932" # Character encoding of string type field
record-length:    2120    # Length of each record
record-separator: "\r\n"  # Carriage return and Line feed

[data]
Item definition omitted

ProjectInsertMessage_RECEIVE.fmt

file-type:        "Fixed" # Fixed-length
text-encoding:    "MS932" # Character encoding of string type field
record-length:    130     # Length of each record
record-separator: "\r\n"  # Carriage return and Line feed

[data]
Item definition omitted

Send/receive process using MessageSender

Point

• Create the request message with SyncMessage.
• To send a message, use MessageSender#sendSync. For How to Use, refer to the linked Javadoc.

SendProjectInsertMessageAction.java

public Result handle(SqlRow inputData, ExecutionContext ctx) {

    // Business process using input data is omitted

    SyncMessage responseMessage = null;
    try {
        responseMessage = MessageSender.sendSync(
            new SyncMessage("ProjectInsertMessage").addDataRecord(inputData));
    } catch (MessagingException e) {
        // Send error
        throw new TransactionAbnormalEnd(100, e, "error.sendServer.fail");
    }

    Map<String, Object> responseDataRecord = responseMessage.getDataRecord();

    // Business process using response data is omitted

    return new Success();
}

Configuration of MessageSender

Point

• When MessageSender is used, configuration of queue name of send and receive destination and storage directory of format definition file are required. Configured by Set the dependent value. For configuration items, see MessageSenderSettings.
• If you want to change the conversion process of sent and received messages, you can change it by defining a class that inherits SyncMessageConvertor in the component definition file and specifying the name of the component in messageSender.DEFAULT.messageConvertorName. For details, see Change the reading and writing of the framework control header(For sending synchronous message).

messaging.properties

messageSender.DEFAULT.messagingProviderName=defaultMessagingProvider
messageSender.DEFAULT.destination=TEST.REQUEST
messageSender.DEFAULT.replyTo=TEST.RESPONSE
messageSender.DEFAULT.retryCount=10
messageSender.DEFAULT.formatDir=format
messageSender.DEFAULT.headerFormatName=HEADER

component definition file

<!-- Load MessageSender settings -->
<config-file file="messaging/messaging.properties"/>

7.6.1.3.4. Receive message with asynchronous response (Receiving asynchronous message)¶

Receive a message that is sent to a specific destination. Block until a message is received or the wait timeout expires.

Content of common protocol header Common protocol header of received message created by the external system

Message ID:	Configuration not required (numbered after sending)
Correlation message ID:
	Configuration not required
Send address:	Logical name of the address
Reply to address:
	Configuration not required
Expiry interval:
	Any

Receiving asynchronous message provides AsyncMessageReceiveAction as a common action to save the received message in the temporary table (message receive table). AsyncMessageReceiveAction is an action class that works with Messaging with MOM.

Only the following deliverables are required to be created using AsyncMessageReceiveAction and the messages can be saved in the table very easily.

• Temporary table to register the messages
• Format definition file that represents the message layout
• INSERT statement (SQL file) for registering a message
• Form class used to register a message

Below is an implementation example based on Example application to receive project information.

Implementation examples

Temporary table to register the messages

Point

• Received messages are saved in a dedicated temporary table for each message type.
• The primary key is the column that stores the ID that uniquely identifies the message. The value to be stored in this column is numbered by the framework using Surrogate Key Numbering.
• The attribute information of the table defines the columns corresponding to each item of the message to be received.
• Define common items (registration user ID, registration date and time, etc.) according to each project architecture.

INS_PROJECT_RECEIVE_MESSAGE

Receive message serial number (PK)	RECEIVED_MESSAGE_SEQUENCE
Project name	PROJECT_NAME
Project type	PROJECT_TYPE
Project classification	PROJECT_CLASS
: (Omitted)
Status	STATUS
Registered user ID	INSERT_USER_ID
Registration date and time	INSERT_DATE

Format definition file

Point

• The file name is <Request ID of received message>_RECEIVE.fmt.

ProjectInsertMessage_RECEIVE.fmt

file-type:        "Fixed" # Fixed-length
text-encoding:    "MS932" # Character encoding of string type field
record-length:    2120    # Length of each record

[userData]
Item definition omitted

SQL file

Point

• The file name is <Request ID of received message>.sql.
• SQL_ID is INSERT_MESSAGE.

ProjectInsertMessage.sql

INSERT_MESSAGE =
INSERT INTO INS_PROJECT_RECEIVE_MESSAGE (
    RECEIVED_MESSAGE_SEQUENCE,
    PROJECT_NAME,
    PROJECT_TYPE,
    PROJECT_CLASS,
    Below omitted

Form class used to register a message

Point

• The file name is <Request ID of received message>Form.
• Define a constructor with two arguments: String, RequestMessage. The meaning of each parameter is as follows.
  • String -> Received message serial number
  • RequestMessage -> Received message

ProjectInsertMessageForm.java

public class ProjectInsertMessageForm {

    /** Received message serial number */
    private String receivedMessageSequence;

    /** Project name */
    private String projectName;

    // Other fields are omitted

    public ProjectInsertMessageForm(
            String receivedMessageSequence, RequestMessage message) {
        this.receivedMessageSequence = receivedMessageSequence;

        DataRecord data = message.getRecordOf("userData");

        projectName = data.getString("projectName");

        // Subsequent process is omitted
    }

    // Accessor is omitted
}

Configuration of AsyncMessageReceiveAction

Point

• When AsyncMessageReceiveAction is used, configuration of format definition file and SQL file location, etc. are required. Configured by adding AsyncMessageReceiveActionSettings to the component definition. For the configuration items, refer to the Javadoc link.

messaging-async-receive-component-configuration.xml

<component name="asyncMessageReceiveActionSettings"
           class="nablarch.fw.messaging.action.AsyncMessageReceiveActionSettings">
  <property name="formClassPackage" value="com.nablarch.example.form" />
  <property name="receivedSequenceFormatter">
    <component class="nablarch.common.idgenerator.formatter.LpadFormatter">
      <property name="length" value="10" />
      <property name="paddingChar" value="0" />
    </component>
  </property>
  <property name="receivedSequenceGenerator" ref="idGenerator" />
  <property name="targetGenerateId" value="9991" />
  <property name="sqlFilePackage" value="com.nablarch.example.sql" />
</component>

Applying AsyncMessageReceiveAction

Point

• To make AsyncMessageReceiveAction work with Messaging with MOM, specify AsyncMessageReceiveAction in the component definition of Request Dispatch Handler.

messaging-async-receive-component-configuration.xml

<component class="nablarch.fw.handler.RequestPathJavaPackageMapping">
  <property name="basePackage"
            value="nablarch.fw.messaging.action.AsyncMessageReceiveAction" />
  <property name="immediate" value="false" />
</component>

7.6.1.3.5. Receive message with synchronous response (Receiving synchronous message)¶

It receives a message sent to a specific destination from a communication destination, and sends a response message to the reply to address configured in the message. At this time, the value of the message ID header of the received message is configured in the correlation message ID header of the response message.

Contents of Common protocol header configured in send message

Message ID:	Configuration not required (numbered after sending)
Correlation message ID:
	Value of message ID header of received message
Send address:	Value of the reply to address header of received message
Reply to address:
	Configuration not required
Expiry interval:
	Any

Receiving synchronous message provides MessagingAction as a template class that performs routine processing. MessagingAction is an action class that works with Messaging with MOM.

Only the following deliverables are required to be created by using MessagingAction.

• Format definition file that represents the message layout
• Process when a message is received and when an error occurs (action class)

Below is an implementation example based on Example application to receive project information.

Implementation examples

Format definition file

Point

• The file name is as follows.

• To receive: <Message request ID>_RECEIVE.fmt
• To send: <Message request ID>_SEND.fmt

ProjectInsertMessage_RECEIVE.fmt

file-type:        "Fixed" # Fixed-length
text-encoding:    "MS932" # Character encoding of string type field
record-length:    2120    # Length of each record
record-separator: "\r\n"  # Carriage return and Line feed

[data]
Item definition omitted

ProjectInsertMessage_SEND.fmt

file-type:        "Fixed" # Fixed-length
text-encoding:    "MS932" # Character encoding of string type field
record-length:    130     # Length of each record
record-separator: "\r\n"  # Carriage return and Line feed

[data]
Item definition omitted

Process when a message is received and when an error occurs (action class)

Point

• Inherit MessagingAction and override the following method.

  • MessagingAction#onReceive
  • MessagingAction#onError

• Create the response message with RequestMessage#reply.

• To retain the contents of request/response message, create a form class corresponding to each.

ProjectInsertMessageAction.java

public class ProjectInsertMessageAction extends MessagingAction {

    @Override
    protected ResponseMessage onReceive(
            RequestMessage request, ExecutionContext context) {
         ProjectInsertMessageForm projectInsertMessageForm
            = BeanUtil.createAndCopy(
                ProjectInsertMessageForm.class, request.getParamMap());

        // Perform the validation process.ApplicationException is thrown when an error is detected.
        ValidatorUtil.validate(projectInsertMessageForm);

        ProjectTemp projectTemp
            = BeanUtil.createAndCopy(
                ProjectTemp.class, projectInsertMessageForm);

        // Data configuration is omitted

        UniversalDao.insert(projectTemp);

        // Returns response data
        ProjectInsertMessageResponseForm resForm = new ProjectInsertMessageResponseForm("success", "");
        return request.reply().addRecord(resForm);
    }

    @Override
    protected ResponseMessage onError(
            Throwable e, RequestMessage request, ExecutionContext context) {

        if (e instanceof InvalidDataFormatException) {
            //Invalid request message data record section layout
            resForm = new ProjectInsertMessageResponseForm("fatal", "invalid layout.");
        } else if (e instanceof ApplicationException) {
            //Request message data record section item validation error
            resForm = new ProjectInsertMessageResponseForm("error.validation", "");
        } else {
            resForm = new ProjectInsertMessageResponseForm("fatal", "unexpected exception.");
        }
        return request.reply().addRecord(resForm);
    }
}

7.6.1.4.1. Change the reading and writing of the framework control header¶

In some cases, reading and writing of the framework control header may require to be changed when the message format is already defined in the external system. To support this, add the implementation in the project. The following shows the support method for each type of send and receive.

For sending asynchronous message

Since writing of the framework control header is performed by the following method, support is provided by overriding the following method.

• AsyncMessageSendAction#createHeaderRecordFormatter
• AsyncMessageSendAction#createHeaderRecord

For sending synchronous message

MessageSender delegates the conversion process to SyncMessageConvertor so the conversion process of sent and received messages can be changed, and this class reads and writes the framework control headers.

Therefore, create a class that inherits SyncMessageConvertor, and specify in the configuration of MessageSender. See MessageSenderSettings for the configuration of MessageSender.

For receiving asynchronous message

The framework control header is read by a class that implements the FwHeaderDefinition interface configured in FwHeaderReader. StandardFwHeaderDefinition is used by default.

Therefore, referring to StandardFwHeaderDefinition, a class that implements the FwHeaderDefinition interface can be created in the project and specified as the FwHeaderReader#fwHeaderDefinition property in the component definition.

For receiving synchronous message

Reading the framework control header is the same as receiving asynchronous message.

Even when writing framework control header, creating the class with the implementation of FwHeaderDefinition interface is the same, but the class created with component definition can be specified in the fwHeaderDefinition property of Message Response Control Handler.