How to configure persistent RM in WSO2 WSAS

WSO2 Web Services Application Server (WSAS) provides with a storage manager in which you can persist messages transmitted between RMS (reliable messaging source) and RMD (reliable messaging destination). By configuring persistent RM, you can prevent messages loses even if the server is down. This post guides you through the steps to configure persistent RM in the latest versions of WSO2 WSAS (WSAS-3.*). I assume you have a basic understanding about WS-RM, if not, please read http://www.infoq.com/articles/fremantle-wsrm-introduction first.
Pre-requisites
Download and install WSO2 WSAS-3.1.1

Persistent implementation of WS-RM will be supported in most of the popular DBMSs. However, I will use MySQL server 5.* in this demonstration. Therefore, install and configure MySQL.

Step 1
As the first step, we need to create the persistent storage which is going to be used in reliable message transmission. Lets create a data base and the necessary table as follows.

mysql>create database SANDESHA_DB;


mysql>use SANDESHA_DB;


mysql>create table wsrm_sender (
message_id varchar(255) not null,
message_context_ref_key varchar(255),
internal_sequence_id varchar(255),
sequence_id varchar(255),
to_address varchar(255),
inbound_sequence_id varchar(255),
send smallint,
sent_count integer,
message_number bigint,
resend smallint,
time_to_send bigint,
message_type integer,
last_message smallint,
inbound_message_number bigint,
transport_available smallint,
flags integer,
primary key (message_id)
);

mysql>create table wsrm_rmd (
sequence_id varchar(255) not null,
to_epr_addr varchar(255),
to_epr blob,
reply_to_epr_addr varchar(255),
reply_to_epr blob,
acks_to_epr_addr varchar(255),
acks_to_epr blob,
rm_version varchar(255),
security_token_data varchar(255),
last_activated_time bigint,
closed smallint,
terminated_flag smallint,
polling_mode smallint,
service_name varchar(255),
flags integer,
reference_message_key varchar(255),
highest_in_message_id varchar(255),
last_in_message_id varchar(255),
server_completed_messages blob,
outof_order_ranges blob,
to_address varchar(255),
outbound_internal_sequence varchar(255),
next_msgno_to_process bigint,
highest_in_message_number bigint,
rmd_flags integer,
primary key (sequence_id)
);

mysql>create table wsrm_rms (
create_seq_msg_id varchar(255) not null,
sequence_id varchar(255),
to_epr_addr varchar(255),
to_epr blob,
reply_to_epr_addr varchar(255),
reply_to_epr blob,
acks_to_epr_addr varchar(255),
acks_to_epr blob,
rm_version varchar(255),
security_token_data varchar(255),
last_activated_time BIGINT,
closed smallint,
terminated_flag smallint,
polling_mode smallint,
service_name varchar(255),
flags integer,
id bigint,
internal_sequence_id varchar(255),
create_sequence_msg_store_key varchar(255),
reference_msg_store_key varchar(255),
last_send_error blob,
highest_out_relates_to varchar(255),
client_completed_messages blob,
transport_to varchar(255),
offered_endpoint varchar(255),
offered_sequence varchar(255),
anonymous_uuid varchar(255),
last_send_error_timestamp bigint,
last_out_message bigint,
highest_out_message_number bigint,
next_message_number bigint,
terminate_added smallint,
timed_out smallint,
sequence_closed_client smallint,
expected_replies bigint,
soap_version integer,
termination_pauser_for_cs smallint,
avoid_auto_termination smallint,
rms_flags integer,
primary key (create_seq_msg_id)
);

mysql>create table wsrm_invoker (
message_context_ref_key varchar(255) not null,
sequence_id varchar(255),
context blob,
msg_no bigint,
flags integer,
PRIMARY KEY (message_context_ref_key)
);

mysql>create table wsrm_msgctx (
ctx_key varchar(255) not null,
ctx blob,
PRIMARY KEY (ctx_key)
);

Note: From WSAS-3.2.0 onwards, the following columns must be added to the wsrm_rms table in addition to the columns specified above.

offered_endpoint_epr_addr varchar(255),
offered_endpoint_epr varchar(255),
reallocated integer,
internalSeqIDOfSeqUsedForReallocation varchar(255),

Step 2

Since we are going to establish a JDBC connection to a MySQL DB, copy mysql jdbc driver to WSAS_HOME/repository/components/lib and restart WSAS
Log in to WSAS management console and click on modules -->list. Click on configure link of the sandesha2 module.
By default, WSO2 WSAS uses an inmemory implementation of WS-RM. You can change it to persistent by changing Storage Manager property. Configure the rest of the persistent storage details as follows.

Connection String - jdbc:mysql://localhost:3306/SANDESHA_DB
Driver Name - com.mysql.jdbc.Driver
Username - your mysql username 
Password - your mysql password

Step 3

Now, you have enabled persistent RM in WSO2 WSAS. However, due to a bug in the latest WSAS releases, you should update the following property in WSAS_HOME/conf/axis2.xml

<parameter name="Sandesha2StorageManager">persistent</parameter>

Make sure to restart WSAS after configuring the above element in axis2.xml

Step 4

Now, you are ready to send requests to a service hosted in WSAS through RM and verify persistent RM. In order to do that, engage sandesha2 module in a service and send series of requests (say 100 requests) using a RM client (I assume you are familiar with sending soap messages with RM headers. I will publish a separate post on complete RM client-server invocation soon). After transmitting 30 requests, shutdown WSAS and restart. You will notice that after the server is restarted, the message transmission will be resumed from the message no: 31

If you have any issues with the steps given above, please drop me a mail.

WS-Reliable Messaging with WSO2 WSAS - II

This is the part two of my previous post on WS-Reliable messaging with WSO2 WSAS - 1
I'd suggest you to follow the steps given in there before proceeding with this.

We looked at how WSO2 WSAS and WSO2 Mercury module are used in one-way (IN-ONLY) message interaction in reliable manner. In this post I'll explain the procedure to do the same with Request Reply(IN-OUT) messaging system using single transport channel.

Step 1

Follow the step 1 - Service Configuration and step 2 - Generate stubs of part one of this post with the service implementation class given below.

package org.wso2.wsas.service;
public class MercuryService{

//Twoway messaging
public int Add(int a, int b){

try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return a+b;
}
}

Note that I inserted a Thread.sleep() in service to wait ~3 seconds before returning results so that we may have enough time to block the transport channel when testing reliable message transmission.

Step 2

If you have followed the step 1 and step 2 exactly as given in part 1 of this post by just replacing service impl. class with the above, you should be able to continue with the client as given below.

Open your IDE and write the following client to invoke service with RM.
Make sure to add jars in WSAS_HOME/lib and the generated client stub jar to your class path in order to compile the client.

public class RequestReplyAnnonClient{

public static void main(String args[])throws AxisFault{

ConfigurationContext cc = ConfigurationContextFactory.createConfigurationContextFromFileSystem
("C:\\RM\\client-repo"+"C:\\RM\\client-repo\\conf\\axis2.xml");

MercuryServiceStub stub = new MercuryServiceStub(cc);
MercuryServiceStub.Add request = new MercuryServiceStub.Add();
request.setA(12);
request.setB(8);

stub._getServiceClient().engageModule("Mercury");

for (int i = 0; i < 29; i++) {
try {
//Two-way invocation
MercuryServiceStub.AddResponse response = stub.Add(request); System.out.println(response.get_return());
} catch (RemoteException e) {
e.printStackTrace();
} try
{
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// Setting the last message stub._getServiceClient().getOptions().setProperty("MercuryLastMessage", Constants.VALUE_TRUE);
try {
MercuryServiceStub.AddResponse response = stub.Add(request); System.out.println(response.get_return());
} catch (RemoteException e) {
e.printStackTrace();
} try
{ Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
In order to have the reliable messaging support at the client, we need to engage Mercury module with "stub._getServiceClient().engageModule("Mercury")". We also need to have ConfigurationContext to define the client repository where the Mercury.mar and axis2 configuration file (axis2.xml) is placed. Create a directory, client-repo and copy WSAS_HOME/repository/modules/mercury.mar to client-repo/modules directory. Also, create a sub-directory conf under client-repo and copy an axis2.xml file ( I have used axis2.xml copied from Axis2-1.4) in to that folder. As in one-way scenario, we need to let Mercury know when we are complete. The simplest way to do this is to mark the LastMessage with stub._getServiceClient().getOptions().setProperty("MercuryLastMessage", Constants.VALUE_TRUE)

Step 3

Compile and run the client. You should configure tcpmon in order to monitor request and response messages.

You will notice that the first message travels through tcpmon contains <wsrm:CreateSequence> as the first first child of soap body. If the transport channel does not get blocked, CreateSequenceResponse message will be returned to client immediately.

You will see that <wsrm:MessageNumber>1</wsrm:MessageNumber> soap header element in the next message where the actual request payload is transmitted. The response of this message may include <wsrm:SequenceAcknowledgement> which denotes the ack to the first message. This will be continued until the last message in which the <wsrm:LastMessage /> as a header element could be seen. (In our test, the message number should be 30)
The end of whole sequence will be denoted by TerminateSequance message.

Lets check whether the reliability is actually provided by WSO2 mercury module when the transport channel is interrupted.

When the 3rd message is sent (Message number 3), click on "Stop" button of Tcpmon listen port. You will see the following in tcpmon.



This mimics the behavior of typical network interruption in data communication. Wait a few seconds and restart the listen port.



As you can see, the response of request message number 3 will be returned when the transport channel is restarted and continues with the rest of the messages. No messages are lost though the channel interruptions are occurred. All 30 messages will be delivered reliably even the network communication issues are occurred during message transmission.

We have looked at one of a basic tests which can be done in order to verify how WSO2 WSAS and mercury module provides reliable message delivery in web service communication. There are more. I'll discuss them in a future post.