How the ESB Publish-Subscribe Channel works

Publish-Subscribe is an  EIP pattern where the sender can sends the message to the different subscribers who subscribes to it. The Publish-Subscribe Channel EIP receives messages from the input channel (publisher) and then splits and transmits them among its subscribers through the output channel.

This blog post explains  how the Publish-Subscribe Channel EIP works.

The message comes from publisher and it directs to the WSO2 ESB. The "Event" mediator resides in proxy service allows you to define set of receivers and redirect the incoming event to the correct event topic.

To understand how this works, follow the steps given below.

Here we use echo service hosted in WSO2 application server to explain this.

1.Create a topic called “pubsub1" in ESB topics and subscribed to it using following end points

http://localhost:9763/services/echo

http://localhost:9767/services/echo

Note : You  need to start two WSO2  Application server instances



2. Create a proxy service in ESB as follows

<?xml version="1.0" encoding="UTF-8"?>

<proxy xmlns="http://ws.apache.org/ns/synapse"

name="pubsub"

transports="https,http"

statistics="disable"

trace="disable"

startOnLoad="true">

<target>

<inSequence>

<log level="full"/>

<event topic="pubsub1"/>

</inSequence>

<faultSequence>

<log level="full">

<property name="MESSAGE" value="Executing default &#34;fault&#34; sequence"/>

<property name="ERROR_CODE" expression="get-property('ERROR_CODE')"/>

<property name="ERROR_MESSAGE" expression="get-property('ERROR_MESSAGE')"/>

</log>

<drop/>

</faultSequence>

</target>

<description/>

</proxy>

Send the following message

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:p="" xmlns:xsd="http://service.carbon.wso2.org/xsd">

<soapenv:Header/>

<soapenv:Body>

<p:echoString>

<in>a</in>

</p:echoString>

</soapenv:Body>

</soapenv:Envelope>

This is how you verify whether the message is subscribed successfully.

Invoke the proxy and go to the following location in application server " Home     > Monitor     > System Statistics".  Verify the request count is changed according to the no of times you invoke the proxy. Check theses in both server instances.


Server Instance 1

Home     > Monitor     > System Statistics
 

 

 Server Instance 2

How the Message Translator EIP Works

If you need to connect two applications, the main impotent point is both applications should use common data type compatible with each other. To facilitate this, some intermediary translation should happen between the sender and receiver. The translator in between  should change the context of the message form one data type to another.

This blog post explains how ESB Message Translator EIP works.

To explain this EIP pattern we use echo service hosted in application server connect with ESB proxy service. In here sender sends the request in one format and the translator transforms that message compatible with the back end service. Here we used the payload factory mediator type “xml “and format the message  coming in to in-sequence compatible with the back end service hosted in application server. The echo service deployed in application  server receives the message compatible with it. 

 Now lets see the how to implement this pattern.

1. Create a proxy service in ESB as follows.

<?xml version="1.0" encoding="UTF-8"?>
<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="translatorProxy"
       transports="https,http"
       statistics="disable"
       trace="disable"
       startOnLoad="true">
   <target>
      <inSequence>
         <log level="full"/>
         <log level="custom">
            <property xmlns:p="http://service.carbon.wso2.org"
                      name="Value"
                      expression="//p:string"/>
         </log>
         <payloadFactory media-type="xml">
            <format>
               <p:echoString xmlns:p="http://echo.services.core.carbon.wso2.org">
                  <in>$1</in>
               </p:echoString>
            </format>
            <args>
               <arg xmlns:p="http://service.carbon.wso2.org"
                    evaluator="xml"
                    expression="//p:string"/>
            </args>
         </payloadFactory>
         <log level="full"/>
         <send>
            <endpoint>
               <address uri="http://localhost:9764/services/echo"/>
            </endpoint>
         </send>
      </inSequence>
      <outSequence>
         <send/>
      </outSequence>
      <faultSequence>
         <log level="full">
            <property name="MESSAGE" value="Executing default &#34;fault&#34; sequence"/>
            <property name="ERROR_CODE" expression="get-property('ERROR_CODE')"/>
            <property name="ERROR_MESSAGE" expression="get-property('ERROR_MESSAGE')"/>
         </log>
         <drop/>
      </faultSequence>
   </target>
   <description/>
</proxy>  

                            

2. Open "SOAPUI"/"Try it" Client and send the below message to the ESB proxy

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:p="http://service.carbon.wso2.org">

<soapenv:Body>
<p:string>testString</p:string>

</soapenv:Body>

</soapenv:Envelope>

3. After sending the request, the payload factory mediator in ESB proxy  converts the request compatible with the receiver. 

The message format compatible with the receiver is as follows

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:p="http://service.carbon.wso2.org">

<soapenv:Body>

<p:echoString xmlns:p="http://echo.services.core.carbon.wso2.org">

<in xmlns="http://ws.apache.org/ns/synapse">testString</in>

</p:echoString>

</soapenv:Body>

</soapenv:Envelope>

How the Message Router EIP can be implemented using WSO2 ESB

Introduction to Message Router.

Message router reads the content of the message, and based on the content of the message it directs to the incoming message to the correct recipient. Message router is useful handling scenarios like when the specific logical function is distributed among several physical systems and we need to deliver the message to the correct service.

The following diagram describes behavior of message router.

Message router by practical example

Now lets have a look on how message router works by a practical scenario.

Assume that we have deployed "Echo" service in WSO2 Application server. Here we need to deliver the incoming message (string or integer) to the different locations. What you need to do is first create a proxy service in WSO2 ESB by giving the end points of the services and WSDL of the echo service deployed in Application server.

Create a proxy service as follows in WSO2 ESB

<?xml version="1.0" encoding="UTF-8"?>
<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="test1"
       transports="https,http"
       statistics="disable"
       trace="disable"
       startOnLoad="true">
   <target>
      <inSequence>
         <log level="full"/>
         <switch source="local-name(/*/*/*)">
            <case regex="echoString">
               <log level="custom">
                  <property name="return" value="String"/>
               </log>
               <send>
                  <endpoint>
                     <address uri="http://localhost:9763/services/echo"/>
                  </endpoint>
               </send>
            </case>
            <case regex="echoInt">
               <log level="custom">
                  <property name="return" value="Int"/>
               </log>
               <send>
                  <endpoint>
                     <address uri="http://localhost:9764/services/echo"/>
                  </endpoint>
               </send>
            </case>
            <default>
               <log level="custom">
                  <property name="return" value="default case"/>
                  <property name="return" expression="local-name(/*/*/*)"/>
               </log>
            </default>
         </switch>
      </inSequence>
      <outSequence>
         <send/>
      </outSequence>
      <faultSequence>
         <log level="full">
            <property name="MESSAGE" value="Executing default 'fault' sequence"/>
            <property name="ERROR_CODE" expression="get-property('ERROR_CODE')"/>
            <property name="ERROR_MESSAGE" expression="get-property('ERROR_MESSAGE')"/>
         </log>
         <drop/>
      </faultSequence>
   </target>
   <publishWSDL uri="http://localhost:9763/services/echo?wsdl"/>
   <description/>
</proxy>

What happens in this proxy service is switch mediator observes the message and filters out the content based on the given xpath  expression. If the filtered content is matched with the case,(string/integer) the message will be routed to the specific endpoint. if the matching condition is not fount the message will be directed to the default case.

Once you invoke the above proxy you will receive a responses as follows.

case 1 - string

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:echo="http://echo.services.core.carbon.wso2.org">
<soapenv:Body>
      <echo:echoString>
         <!--Optional:-->
         <in>chathurika</in>
      </echo:echoString>
   </soapenv:Body>
</soapenv:Envelope>

[2014-04-06 00:01:46,824]  INFO - LogMediator return = String

case 2 - integer

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:echo="http://echo.services.core.carbon.wso2.org">
<soapenv:Body>
      <echo:echoInt>
         <!--Optional:-->
         <in>123</in>
      </echo:echoInt>
   </soapenv:Body>
</soapenv:Envelope>

[2014-04-06 00:05:28,934]  INFO - LogMediator return = Int

Case - default

<soapenv:Envelope xmlns:soapenv="http://www.w3.org/2003/05/soap-envelope"><soapenv:Body><p:echoStringArrays xmlns:p="http://echo.services.core.carbon.wso2.org">
<!--0 or more occurrences--><a>aaa</a>
<!--0 or more occurrences--><b>bbbb</b>
<!--0 to 1 occurrence--><c>vvvvv</c>
</p:echoStringArrays></soapenv:Body>
</soapenv:Envelope>

[2014-04-06 00:33:55,953]  INFO - LogMediator return = default case, return = echoStringArrays