MQTT Transport for Axis2/Synapse

MQTT is an application layer broker based pub/sub protocol built on top of TCP/IP. It is very lightweight and hence ideal for usage in network/memory/computing power constrained environments. If I quote the MQTT spec, the protocol features,

1. Pub/Sub message pattern
2. Small transport overhead, fixed length header being only 2 bytes
3. Supports three QOS for message delivery, at-least once, exactly once and at-most once
4. etc.

The ability to operate under constrained environment make this protocol an ideal candidate for embedded device communication/pervasive computing.  However smaller footprint is always better. Why not MQTT in enterprise integration ? :)

There are number of MQTT broker implementations out there (you can find the original spec here). Right now, MQTT transport is undergoing OASIS standardization process. Mosquitto is one such broker implementation of MQTT. Eclipse Paho project provides MQTT client library in Java. 

Apache Synapse is  a popular open source enterprise service bus (ESB). Synapse mediation engine relies on Axis2 transport framework to receive messages. Axis2/Synapse  supports many transports out  of the box. I set out to implement a MQTT transport for Axis2/Synapse so that MQTT transport is available for EIP scenarios/etc with Synapse ESB.

The sample message flow,

Setting up the environment and running the code,

1. Download the MQTT broker from Mosquitto.org and install

2. Download Apache Synapse and install. Due to some API compatibility reasons, the code only works with trunk version of Synapse (1.7.0-SNAPSHOT) as of this writing.

3. Deploy the provided proxy service configuration 'SampleProxy.xml'
The provided proxy service read from a Mosquitto broker topic and writes to a publicly hosted MQTT broker topic.

Next configure the MQTT transport definition in axis2.xml transports section. Here we configure the transport listener and transport sender.

        
                localhost
                1883
                esb.test.listener
                esb.test2
        
    


    

simply copy replace the original axis2.xml file found in synapse distribution with provided axis2.xml.
Now add the eclipse Paho MQTT client jar and the axis2-mqtt transport jar to synapse classPath. That is to $synapse_home/lib/

4. Start Synapse and Mosquitto broker

5. Publish message to topic1 using provided client. Edit the endpoint addresses and payload accordingly.

Once you publish the message to the topic, the listening ESB transport will get notified and proxy service will write the message to the target endpoint.

6. Read/Observer the output in topic2 using the provided receiver. The online hosted MQTT broker is configured as the target endpoint. The http bridge allows you to see the currently published topics to the broker.

I have only demonstrated, MQTT -- > MQTT scenario. The Axis2 transport framework allows you to transport switch messages. Hence X --> MQTT / MQTT--> X is possible. (X is a Axis2 transport). The code is very much in its initial state. The next step would be to improve this and contributing to Axis2 project.

Dealing with Occasional Message Bursts : Message Broker Use Case

Recently I came across the following requirement,

An organization has a message delivery network and an application that makes use of those messages (process them). The delivery network directly communicates with the message processing application. Things seems to be working fine on a regular day. However if the load(incurred by the delivery network) is high, some of the messages may get lost before they get processed (overflow of applications' task queue).

Message brokers are designed to cater the exact scenario. Enterprises rely heavily on their legacy systems and changing all existing integration points may not be the right solution and will never work. In the above scenario, we can place a message broker in between the message producer and message consumer. By introducing a message broker we,

1. Decouple producer/consumer in time and space (two parties need not to be functioning at the same time and the don't have to know each others existence )   

2. Reliable message processing/delivery by means of persistence and transactions.

I implemented the POC scenario using WSO2 ESB and WSO2 MB. Please find the configuration files and source code in the this location. ESB used for its JMS transport support (you don't have to change the producer code).

How to use POC config/code;

1. replace the the original axis2.xml and jndi.properties file found in WSO2ESB distribution with provided ones. Here I have enabled JMS sender in axis2.xml and configured connection factory params in jndi.properties file.

2. Create a proxy service in WSO2 ESB using provided StockQuoteProxy.xml. This proxy will accept HTTP request and write them in to a MB queue.

3. Create a new queue named 'StockQuotesQueue' in WSO2MB: this is the queue we are writing to and reading from.

4. Start server and run the servers, send some soap messages to proxy services, so that those get written to MB. 

5. Run the Receiver to retrieve messages from MB queue.

please configure ports and hosts accordingly. Refer WSO2 MB documentation as well.

Fine Grained XACML Based Authorization With FuseSource ESB and WSO2 IS

Authorization - The bad way

In a typical organization that makes use of Service Oriented Architecture (fully/partially) to drive their IT requirements, chances are more to find tens/hundreds of services scattered across departments. If the legacy services are works fine for the organization, changing them is not a wise decision. However one always has to consider the maintainability of the system in the long run. If the services are coherent, that is, if they are doing only what they are supposed to do(known as only the business logic) then the changes are that, they may never wanted to be modified ever.
However we are not living in a ideal world. If i take the authorization as an example, I have seen people burning down authorization logic within their service implementation. This is a bad practice, since.

1. If the authorization mechanism(verifying against a JDBC user/permission store) changes you have to change the service implementation or introduce a hack to cover it up.
2. If the authorization policy(from now on we allow clients from X organization with security lever Y to access the service) changes then you have to change the actual service.

In simple words its not future proof. If the organization wants to monitor the authorization activities they you might end up modifying all your existing services.


What is XACML 

With XACML we can declaratively define our authorization policy using XML. Once you come up with the policy XACML engine evaluates the incoming message against the store XACML policy and spits the final decision (allow/deny). Since XACML is accepted by OASIS and is the de-facto standard which is supported by many industry leaders, there is little chance for vendor locking in the long run.

Sample Scenario

   
1. Clients sends in the request to access back-end service/resource. Here the request is directed to Fuse ESB proxy service. 
2. The proxy service intercepts(acts as the Policy Enforcement Point - PEP) the message and the entitlement bean extracts message properties and create the XACML request, which then forwarded to XACML engine. (Here we are using the XACML engine provided by WSO2 Identity Server).
3. Identity server validates (acts as the Policy Decision Point- PDP) the request against its stored XACML policy/policies and respond back with the decision.
4. Based on the received decision proxy service either forward the original client request to actual backend service or sends and not-authorized fault message to the client.
5. In the former case in the step 4, client receives the response from the actual back-end.

Note: The actual implementation (code) setup and the execution will be explained in a later post.

POST variables get dropped while going through the WSO2 ESB proxy ... ?

This is more of a small note in case I forget what i did while trying things out. First of all I created a ESB proxy for a back-end service that accepts post request and does some manipulation on the received request.

Once you create a ESB proxy by giving the above REST service as the service endpoint , the InSequence element of the proxy.xml will look like this.

<inSequence>
         <send>
            <endpoint name="endpoint_urn_uuid_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx">
               <address uri="http://localhost/restServiceEndPoint" format="rest" />
            </endpoint>
         </send>
      </inSequence>

The problem is the configuration drops the POST variables while building the SOAP envelope. Some debugging in to the Axis2 code revealed that the transport is attaching the incoming variable list to the Axis2MessageContext, but the attached data structure is MultipleEntryHashMap(It removes the value from the data-structure once you access it) which is a Axis2 implementation. However I couldnot access the the stored variables in the data structure within a Custom mediator, (the data has been already consumed at that point, may be by once of the dispatchers.)

So the other option was to use binary relay. It worked fine for me. Here are the steps I followed to get it working. First you have to enable the binary relay builder / Formatter for the relevant application-content type (In my case XForm-URL-encoded).

Then remove the format="rest"  attribute from the address element (InSequence of proxy.xml)

Thats it, the binary relay builder wraps your request as it is and expand it back when it sends back to to actual service.