Thursday, November 22, 2012

Pick Payload Factory from registry in WSO2 ESB

Payload-factory mediator does is transforming message content which is also known as payload, in between client and the actual server. Incoming messages coming from the client can be transformed in to a format expected by the server by using payload-factory mediator. Then again response message can be formatted in a way that client can understand.

Payload-factory mediator has two elements namely format and args. Currently these two elements of payload-factory are defined inline. With my implementation I provide an option to pick the configuration from registry when format element is defined. What I did was, make available two options to define the format of payload factory. They are “Define inline” and pick from registry.

This project consists of mainly two parts which are back-end implementation and User Interface (UI) implementation. I started the project with back-end implementation. There were PayloadFacoryMeadiator class, PayloadFacoryMeadiatorSerializer class and PayloadFacoryMeadiatorFactory class. When payload factory is created PayloadFacoryMeadiatorSerializer serialize the payload-factory as an OMElement. When payload-factory is loaded back OMElement will be extracted by PayloadFacoryMeadiatorFactory.

When payload-factory is configured to get the format from registry, a key should be provided in the format tag. This key is the location of the format configuration file in the registry. The key is saved in PayloadFacoryMeadiator as an attribute. When transforming client’s message to the server expected format, if format is dynamic (pick from registry), format should be taken from MessageContext inside mediate method. Below figure shows how format is picked from MessageContext. Here synCtx means an instance of Message Context.

Below is the payload-factory configuration for ESB synapse configuration when format is dynamic. Here “conf” is referred to as configuration registry. Configuration of the format, dynamic_payload_1.xml, is saved inside a directory called payloadFactory in configuration directory. Configuration file’s location is given as the key of the format as described above. “args” tag defines the arguments passed to the payload-factory mediator.

<format key="conf:/payloadFactory/dynamic_payload_1.xml"/>
<arg xmlns:ns="http://org.apache.synapse/xsd"

Below figure shows the User Interface which I created for payload-factory.

Monday, November 12, 2012

Supporting Hierarchical Proxy Services

When I was in WSO2 ESB team, I got a change to contribute to ESB product in WSO2. I did an improvement to WSO2 ESB proxy services. Proxy service is an intermediate component between client program and the actual server which the client is seeking for. When a proxy service is deployed at that moment ESB save the configuration file inside ESB_HOME/repository/deployment/server/synapse-configs/default/proxy-services/ directory where all the proxy services are saved. If we deploy a proxy service named “foo” that proxy service will be saved in ESB_HOME/repository/deployment/server/synapse-configs/default/proxy-services/ as foo.xml. The service URL for that proxy service is 8280:services/foo.

Since all the proxy services are deployed under the same directory, it is very hard to manage the services for a large organization when the number of services becomes large, because they have to agree on service name before deploying them. And also supporting multiple versions of the same service is not possible without changing the service name in the services.xml file.

Therefore it was one of the requirements of ESB team to have hierarchical proxy services in WSO2 ESB. My task was to implement hierarchical proxy service in WSO2 ESB. When proxy service is created other than the name, there is an option to give service hierarchy also, but service hierarchy is an optional requirement. If service hierarchy is not given the proxy service XML file will be saved in the default directory. Below figure shows a screen shot of User interface which I created in order to add service hierarchy.

I added a new property called serviceHierarchy to the proxy service configuration when a proxy service is added. Below is the proxy service configuration for adding a proxy service named foo and service hierarchy “a/b/c/”.

     <proxy name="foo"
          transports="https http"
            <address uri="http://localhost:9000/services/SimpleStockQuoteService"/>

After deploying this proxy service, proxy configuration XML file will be saved in ESB_HOME/repository/deployment/server/synapse-configs/default/proxy-services/a/b/c/ directory. If the proxy service is removed from the ESB only the file will be deleted. The directory structure will be remained unchanged, because there will be other proxy files inside those directories.

I mainly did changes in synapse core of the ESB package. When proxy service is created ProxyserviceSerializer class serialize the proxy service as an OMElement. I added “serviceHierarchy” as an attribute of the proxy service. When ESB server starts proxy service is loaded from the OMElement by ProxyserviceFactory. Below figure shows what I described above.

Getting Started Message Broker with C#

During my internship at WSO2 I got a chance to contribute to WSO2 Message Broker (MB) by implementing a C# client for WSO2 MB.

WSO2 Message Broker didn’t have any samples in order to connect to WSO2 MB from a C# client. Since WSO2 MB gets the core support from Apache Qpid, it should be possible to connect to WSO2 MB from a C# or .Net client because C# client can connect to Apache Qpid. Apache Qpid is a cross-platform Enterprise Messaging system which implements the Advanced Message Queuing Protocol (AMQP), providing message brokers written in C++ and Java, along with clients for C++, Java JMS, .Net, Python, and Ruby. Even though there is a possibility to get the thing done, no one has tried out it with WSO2 MB.

The rest of this article will describe how to send a message to a queue in WSO2 MB using a C# client and consume that message using a C# client.

WSO2 Message Broker supports JMS, WS-Eventing, Amazon SQS and AMQP. By using AMQP we can create a .Net program to send a message to a queue in MB and retrieve that message from a .Net client or a Java client. In the rest of this tutorial I'm going to explain how to implement a .Net client to send messages to a queue and retrieve those messages using a .Net client and a Java client. When you run these programs, first run the Consumer program and then the Producer program.

Exchanges Queues and bindings in AMQP

Like any other messaging protocol AMQP also uses producers and consumers when sending and receiving messages. The responsibility of the producer is to produce the message and send it to the destination. The consumer consumes that message and processes it. It is the responsibility of the message broker to deliver the message which is produced by the producer, to the consumer.

AMQP uses two components in order to complete this task. They are exchanges and queues. Producer produces messages and delivers to the exchange. Consumer consumes messages from the queue. We use bindings to connect these two components each other. Publisher and the consumer are recognized each other via the exchange name. Usually either the publisher or the consumer create the exchange and make the name of the exchange public. The Queue should be attached to the exchange after declaring the queue. Then broker has to match the messages received by the exchange to the queue. That is done using bindings. Typically queue binding is happening at the client's side. The following diagram explains what I described above.

In order to run this code sample, you need to download and add RabbitMQ.Client.dll file as a reference in your .net project. You can download that dll file from this website

.Net Publisher:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using RabbitMQ.Client;------------------------------------------------------------------------1
namespace RabbitMQ
    class Publisher
        static void Main(string[] args)
            Publisher p=new Publisher();
                p.PublishMessage("This is a Test "+i);
                Console.WriteLine("Sent Message "+i);
             public void PublishMessage(string message)
//Setup the connection with the message broker
            ConnectionFactory factory = new ConnectionFactory();-------------------------------2
            IProtocol protocol = Protocols.AMQP_0_8_QPID;
            factory.VirtualHost = "/carbon";
            factory.UserName = "admin";
            factory.Password = "admin";
            factory.HostName = "localhost";
            factory.Port = 5672;
            factory.Protocol = protocol;
                using (IConnection conn = factory.CreateConnection())
                using (IModel ch = conn.CreateModel())
                 //Declare the exchange for the publisher.Here the exchange type is direct.
ch.ExchangeDeclare("", "direct");--------------------------------------3
                 //Publish the message
        ch.BasicPublish("", "test-queue", null, Encoding.UTF8.GetBytes(message));--------4


import the RabbitMQ.Client.dll file to the program.
Create a connection to the MB. Here I have specified the VirtualHost as “carbon”. If it is not specified it will go the default VirtualHost, “test”.
Declare the exchange type for the publisher. The queue is interested in messages from this exchange. Here the exchange type is direct. That means a message will go to the queues only if the binding key exactly matches the routing key of the message.
Publish the message. Here we need to specify the exchange type and the routing key. In this example the routing key is similar to the queue name which is “test-queue”. The next parameter is IbasicProperties. I kept it as null for this scenario. Last parameter is the message body.


This consumer program fetches messages from the queue and display them in the console. In order to run this code sample, you need to download and add RabbitMQ.Client.dll file as a reference in your .net project as mentioned above.

 using System;

using System.Collections.Generic;
using System.Linq;
using System.Text;
using RabbitMQ.Client;------------------------------------------5

namespace QueueConsumer

    class QueueConsumer
        static void Main(string[] args)
            QueueConsumer qConsumer = new QueueConsumer();

        public void getMessage()
    //Setup the connection with the message broker

            ConnectionFactory factory = new ConnectionFactory();-----------6
            IProtocol protocol = Protocols.AMQP_0_8_QPID;
            factory.VirtualHost = "/carbon";
            factory.UserName = "admin";
            factory.Password = "admin";
            factory.HostName = "localhost";
            factory.Port = 5672;
            factory.Protocol = protocol;
            using (IConnection conn = factory.CreateConnection())
                using (IModel ch = conn.CreateModel())


//Declare a queue to retrieve messages.

                    ch.QueueDeclare("test-queue", true, false, false, null);----7       

 //Create the binding between queue and the exchance

                    ch.QueueBind("test-queue", "", "test-queue");-----8
                    QueueingBasicConsumer consumer = new
                    while (true)-------------------------------------11
                            RabbitMQ.Client.Events.BasicDeliverEventArgs e =
                            byte[] body = e.Body;
                            string message = Encoding.UTF8.GetString(body);
                        catch (OperationCanceledException e)
1. Import the dll file as mentioned above.

2. Create a connection with the message broker.

3. Declare a queue to receive messages. The first parameter is the queue name. The second parameter is durable. I kept it as true. The newt two parameters are exclusive and auto-delete. I kept it as false. The last parameter is IDictionary arguments. It is null here.

4. HereI create the binding for the queue and the exchange. Simply a binding is a relationship between an exchange and a queue. The first parameter is the queue that we want to bind. The next parameter is the exchange that we are binding with the queue and the last parameter is the routing key. Here the routing key is similar to the queue name.

5. Create a consumer to consume the messages from the queue.
6. Gets the message from the queue using the consumer created in the previous step. The first parameter is the name of the queue. I gave the name as “test-queue”. The next parameter is noAck. I kept it as false. For the next parameter I’m passing the consumer object which I created in the previous step.

7. In this infinite while loop, we listen for messages, decoding each one, and printing its body in the console.

The message which I published by the C# client can be consumed by an ordinary Java client also.

Thursday, July 5, 2012

Java Message Service API

Before learn about JMS API lets take a look at what is meant by messaging. Messaging is used to communicate among software component or software applications. A component should send the message to a destination and the recipient component can get the message from that destination. The communication done by messaging is loosely coupled and asynchronous. That is how messaging differs from remote method invocations and polling systems.

Now I'm going to explain what JMS API is by having a better understanding about messaging. Java Message Service API allows component or applications to create, send, receive and read messages. This API is designed by Sun and several partner companies. JMS API communication is loosely coupled and enable asynchronous and reliable (The JMS API can ensure that a message is delivered once and only once) communication service.

JMS consists with four main parts.
  • JMS provider -This implements the JMS interface and provides other administrative and control features
  • JMS client – programs or components written in Java. These can produce and consume messages.
  • Messages – These are communication objects
  • Administered Objects – These are preconfigured JMS objects created by an administrator for clients' usage. e.g. Destination and Connection Factories.

JMS API supports both
  • point to point – Use a queue and there is only one consumer
  • publish/subscribe message domains.
    • There can be multiple consumers
    • messages are based on topics
    • subscribers receive messages which were published after their subscription
    • subscribers must continue to be active in order to consume messages.(JMS API relaxes this timing dependency up to certain extend by allowing subscribers to create durable subscriptions, which receive messages sent while the subscribers are not active)

Version 1.1 of JMS API allows you to use the same code to both domains while message destination remains domain specific. the behavior of the application will depend in part on whether you are using a queue or a topic.

Message Consumption

JMS API provides both synchronous and asynchronous transitions.

  • Synchronous- Subscriber or the receiver explicitly fetches the message by calling the receive method. Receive method will wait until the message arrives or expires certain time limit.
  • Asynchronous- Client can register a message listener with a consumer.Whenever a message arrives at the destination, the JMS provider delivers the message by calling the listener's onMessage method, which acts on the contents of the message.

JMS Programming Components

  • Connection factory creates the connection
    • Connection factory is always a component of ConnectionFactory, QueueConnectionFactory or TopicConnectionFactory.
    • Connection provides an open TCP/IP connection between client and the provider.
  • connection creates the session.
    • One connection can create one or more sessions.
    • Session is a single threaded context to create and consume messages.
  • Session creates the Message Producer, Message Consumer and TextMessages.
  • Message producer creates the message and send it to the destination
  • Message consumer receive messages from the destination
    • receive method is used to receive messages synchronously.

In PTP destination is the queue.
In pub/sub destination is the topic.

How to run QPid WCF samples

 Recently I had to work in a project where I need to connect to Qpid server using a C# client. Before start the project I wanted to try out some samples in Qpid in order to get familiar with Qpid and WCF binding. There I had to face lot of difficulties since lack of gudelines provided by Qpid. Therefore I thought of writing down step by step guidlines to run Qpid WCF samples.

  • Download QPid from
  • Create an environment variable called QPID_ROOT for qpid base directory.(eg:C:\qpid\qpid-0.16)
  • Installl Visual Studio 2008.
  • Install Microsoft Visual Studio 2008 Service Pack 1
  • Install Boost 1.47.
    •     Go to the boost root directory.
    •     Run bootstrap.bat
    •     Run b2 toolset=msvc --build-type=complete stage -j<# cores>
  • Include the location of the Boost library (e.g. %BOOST_ROOT%\lib) in your PATH environment variable.
  • Create an environment variable called BOOST_ROOT for the root of the boost directory.
  • Install SDK 3.5.
  • Install NUnit.
  • Build Qpid cpp

    •     Build CMake.
    •     Build python
    •     Build Ruby.
    •     Ensure that all the build tools are available on your path(Eg:PATH=C:\python25;)
    •     From a command prompt:
      •     # cd qpid\cpp
      •     # cmake -i -G "Visual Studio 9 2008" 
    •     Go to \qpid-0.16\cpp. Open the qpid-cpp.sln solution, select Debug or Release, and build.


  • Create an environmentvariable called QPID_BUILD_ROOT and store the path to the Qpid build directory in it.(Eg:C:\qpid\qpid-0.16\cpp)
  • Open VS 2008 and build(by setting the platform target as x86)

 Building and executing samples


1.Copy the dlls Apache.Qpid.Channel.dll and Apache.Qpid.Interop.dll from %QPID_ROOT%\wcf\src\Apache\Qpid\Channel\bin\x86\Debug
  to the %QPID_ROOT%\wcf\samples\Channel\WCFToWCFDirect folder.

2.Build the solution WCFToWCFDirect.sln.(by setting the platform target as x86))

3.Copy qpidclientd.dll, qpidcommond.dll and qpidtypesd.dll from %QPID_ROOT%\cpp\src\Debug to bin\x86\Debug of each of the projects.
 These dlls are needed at runtime.

4.Copy qpidclientd.dll, qpidcommond.dll and qpidtypesd.dll to %QPID_ROOT%\wcf\tools\QCreate\Debug folder.

5.Start the qpid broker from the qpid build folder e.g. %QPID_ROOT%\cpp\src\Debug.
        command:qpidd.exe  qpidd.exe --data-dir=<to place where it need to be> --auth no
            (Eg:qpidd.exe --data-dir=C:\qpid\qpid-0.16\cpp\src\Debug auth no)

6.Create queue required using the QCreate tool located at
  %QPID_ROOT%\wcf\tools\QCreate\Debug. The syntax is QCreate %QPID_ROOT%. For
  this sample you should do

      QCreate routing_key message_queue
7. Start Service.exe from(Open the cmd as administrator)

8. Start Client.exe from