Publish and Subscribe Scenario

Step1:-
Create a flow like below Publish Messageflow
MQInputNode--->ComputeNode----->Publication Node

Configuration on Nodes:-
MQInputNode:-
Queuename: PublishQueue
Input Message Parsing: XMLNSC

ComputeNode:-

   SET OutputRoot.MQMD.Format =  'MQFMT_RF_HEADER_2';
 SET OutputRoot.MQRFH2.(MQRFH2.Field)version = 2;
 SET OutputRoot.MQRFH2.(MQRFH2.Field)Format = 'MQSTR';
 SET OutputRoot.MQRFH2.(MQRFH2.Field)NameValuCCSID=1208;
 SET OutputRoot.MQRFH2.psc.Command='Publish';
 SET OutputRoot.MQRFH2.psc.Topic = InputRoot.XMLNSC.publish.topic;
 Declare PTR REFERENCE to OutputRoot.MQRFH2;
 Detach PTR;
 attach PTR TO OutputRoot.MQMD AS NEXTSIBLING ;
 SET OutputRoot.XMLNSC.data.info = InputRoot.XMLNSC.publish.info;

Publication Node:-
No configuration needed.

Test Input:-
 <publish><topic>Bhanu</topic><info>Sample</info></publish> Put this Message in  PublishQueue
to publish

Step2:-
Create a flow like below Subscribe Messageflow
MQInputNode--->ComputeNode----->MQOutput Node

Configuration on Nodes:-
MQInputNode:-
Queuename: Subscription
Input Message Parsing: XMLNSC

ComputeNode:-

   SET OutputRoot.MQMD.Format =  'MQFMT_RF_HEADER_2';
 SET OutputRoot.MQRFH2.(MQRFH2.Field)version = 2;
 SET OutputRoot.MQRFH2.(MQRFH2.Field)Format = 'MQSTR';
 SET OutputRoot.MQRFH2.(MQRFH2.Field)NameValuCCSID=1208;
 SET OutputRoot.MQRFH2.psc.Command ='RegSub';
 SET OutputRoot.MQRFH2.psc.Topic = InputRoot.XMLNSC.subscribe.topic;
 SET OutputRoot.MQRFH2.psc.QName = 'RegSub';
 SET OutputRoot.MQRFH2.psc.QMgrName ='TEST';
DECLARE PTR REFERENCE TO OutputRoot.MQRFH2;
 Detach PTR;
 attach PTR TO OutputRoot.MQMD AS NEXTSIBLING ;

MQOutput Node:-
Queuename: SYSTEM.BROKER.CONTROL.QUEUE 

Test Input:-
<subscribe><topic>Bhanu</topic></subscribe> Put this Message in  Subscription
to Subscribe.

Objects to be Created:-

Create PublishQueue,Subscription,RegSub as three queues in MQ for testing

Direction to Test:-

First put the message in Subscription and then in the  PublishQueue, then you can see the message in RegSub queue due to publish and subscribe properties configured in message flow.

Global Cache PUT & GET

Global Cache

IBM Integration Bus provides an out of the box caching mechanism based on WebSphere eXtreme Scale.WebSphere eXtreme Scale provides a scalable, in-memory data grid. The data grid dynamically caches, partitions, replicates, and manages data across multiple servers.

This cache can be used to store reference data that are regularly accessed or to hold a routing table.

The cache is not enabled by default and it is really easy to enable it: the default configuration is activated by setting a configuration parameters through the IBM Explorer administration tool.
To activate the cache across different Integration Node instances, a XML configuration file (templates are provided) has to be defined.

Specialized skills in extremes scale is not necessary in order to use the cache. There is however two important cache components that may be good to know

Catalog servers

component that is embedded in an integration server and that controls placement of data and monitors the health of containers. You must have at least one catalog server in your global cache.

Container servers

component that is embedded in the integration server that holds a subset of the cache data. Between them, all container servers in the global cache host all of the cache data at least once. If more than one container exists, the default cache policy ensures that all data is replicated at least once. In this way, the global cache can cope with the loss of container servers without losing data.

Map

A data structure that maps keys to values. One map is the default map, but the global cache can have several maps.

Each execution group can host a WebSphere XS catalog server, container server, or both. Additionally, each execution group can make a client connection to the cache for use by message flows. The global cache works out of the box, with default settings, and no configuration -- you just switch it on! You do not need to install WebSphere XS alongside the broker, or any other additional components or products.

Scenario:-

Put And Get Key,Values in Global Cache

Write the below code in JCN

MbMessage outMessage = new MbMessage(inMessage);

                                    outAssembly = new MbMessageAssembly(inAssembly, outMessage);

// ----------------------------------------------------------

            // Add user code below

MbGlobalMap map = MbGlobalMap.getGlobalMap();

           

MbElement inParser = inAssembly.getMessage().getRootElement().getLastChild();

String key = "";

if(inParser.getFirstElementByPath("Input/key") != null)

{

key = inParser.getFirstElementByPath("Input/key").getValue().toString();

}

                                                     

MbElement outParser = outAssembly.getMessage().getRootElement().getLastChild ();

outParser.getLastChild().detach();

MbElement result = outParser.createElementAsLastChild(MbElement.TYPE_NAME, "Result", null);

                                   

if(!map.containsKey(key))

{

String value = "";

if(inParser.getFirstElementByPath("Input/value") != null)

      {

value = inParser.getFirstElementByPath("Input/value").getValue().toString();

      }

      map.put(key, value);

result.createElementAsLastChild(MbElement.TYPE_NAME_VALUE, "status", "Key = "+key + "inserted Successful in Global Cache");

      }

      else

      {

      Object value = map.get(key);             

                                          result.createElementAsLastChild(MbElement.TYPE_NAME_VALUE, "Key", key);

result.createElementAsLastChild(MbElement.TYPE_NAME_VALUE, "Value", value);

                                    }

Input Messages:-

//<Input><key>Bhanu</key><value>2032</value></Input>

//<Input><key>Key1</key></Input>

Set below Properties in MQ:-

Now the obtained OutPut Structure will look like:-

Global Cache with Grids Code:-

MbGlobalMap map = MbGlobalMap.getGlobalMap("grid1");

MbGlobalMap mapWithGrid2 = MbGlobalMap.getGlobalMap("grid2");

MbElement inParser = inAssembly.getMessage().getRootElement().getLastChild();

String key = "", value = "";

if(inParser.getFirstElementByPath("Input/key") != null)

{

key = inParser.getFirstElementByPath("Input/key").getValue().toString();

}

if(inParser.getFirstElementByPath("Input/value") != null)

{

value = inParser.getFirstElementByPath("Input/value").getValue().toString();

}

map.put(key, value);

mapWithGrid2.put(key, value);

Object dataFromGrid1 = map.get(key);

Object dataFromGrid2 = mapWithGrid2.get(key);

Backup Of Input Message

Step1:
Create a Message flow like below
MQInputNode---->JavaComputeNode----->MQOutputNode

Step2:
Write the below code in JCN

//Make a modifiable MbMessage based on the input message passed in on the inAssembly.

MbMessage InMessage = new MbMessage(inAssembly.getMessage());

//Now partially parse the input message one record at a time.
MbElement inputRootElement = InMessage.getRootElement();//root
MbElement inputPropertiesElement = inputRootElement.getFirstElementByPath("Properties");//properties
MbElement inputMQMDElement = inputRootElement.getFirstElementByPath("MQMD");//mqmd
MbElement inputXMLNSCElement = inputRootElement.getFirstElementByPath("XMLNSC");//xmlnsc

MbElement inputXMLNSCRootTagElement = inputXMLNSCElement.getFirstChild(); 
//Child(Root tag of Input

 // Create a new output message for the record that we are going to be propagate.
 MbMessage outputMessage = new MbMessage();
outAssembly = new MbMessageAssembly(inAssembly, outputMessage);

//Create new parsers folders in the output message.
 MbElement outputRootElement = outputMessage.getRootElement();//root
 MbElement outputPropertiesElement = outputRootElement.createElementAsLastChild(inputPropertiesElement.getParserClassName());//properties
MbElement outputMQMDElement = outputRootElement.createElementAsLastChild(inputMQMDElement.getParserClassName());//mqmd
MbElement outputXMLNSCElement = outputRootElement.createElementAsLastChild(inputXMLNSCElement.getParserClassName());   //xmlnsc
MbElement BACKUP=outputMessage.getRootElement().createElementAsLastChild(MbElement.TYPE_NAME, "BackUp", null);//New tag in xmlnsc
 
//Create the root tag in the output XMLNSC folder that will be used for this output record.
//  MbElement outputXMLNSCRootTag = inputXMLNSCRootTagElement.createElementAsLastChild(MbElement.TYPE_NAME, "BackUp", null);

//MbElement Payload = outputMessage.getRootElement().getLastChild();
// outputXMLNSCRootTag.addAsFirstChild(outputXMLNSCElement.copy());
 
//Copy the Properties Folder, MQMD header, and the current record.
 outputPropertiesElement.copyElementTree(inputPropertiesElement);//Properties
 
outputMQMDElement.copyElementTree(inputMQMDElement);//mqmd
 
 outputXMLNSCElement.copyElementTree(inputXMLNSCElement); //xmlnsc
 
 BACKUP.copyElementTree(inputXMLNSCElement);//backup
//  MbElement Payload = outputMessage.getRootElement().getLastChild();
// outputRootElement.createElementAsLastChild(MbElement.TYPE_NAME_VALUE, "BackUp",Payload.copy().getValue());  
 
//Propagate this message, requesting that the output message assembly be cleared after propagation.
out.propagate(outAssembly);

Step3:-
Place the below test Message
<Details><Summary><Name>Bhanu</Name><Designation>Technical Lead</Designation><Domain>EAI</Domain><Location>Vishakapatnam</Location></Summary></Details>

Observe the Obtained Output as below

Exception_Email_Notification

Flow:-

MQInput Node Properties:-

Email Ouput Node Properties 

ParseExceptionMessage:-

CREATE COMPUTE MODULE ExceptionHandling_Compute
    CREATE FUNCTION Main() RETURNS BOOLEAN
    BEGIN
         CALL CopyMessageHeaders();
        -- CALL CopyEntireMessage();
       
       
        DECLARE ptrException REFERENCE TO InputExceptionList.*[1];
             DECLARE messageNumber,eText,description,Loc,flowname CHARACTER;
       
         WHILE LASTMOVE(ptrException) DO
            -- store the current values for the error number and text
            IF ptrException.Number IS NOT NULL THEN
                 SET messageNumber = ptrException.Number;
               
                 IF ptrException.Text <> '' THEN
                     IF description<> '' THEN
                        SET description = description || ',' || ' '||ptrException.Text;
                    ELSE
                         SET description = ptrException.Text;
                    END IF;
                END IF;
               
              IF LENGTH(ptrException.Label)>0 THEN
                      SET Loc=ptrException.Label;   
              END IF;
              END IF;
              -- now move to the last child which should be the next exceptionlist
            MOVE ptrException LASTCHILD;
        END WHILE;
       
        SET description = description ||': ' ||ptrException;
        set eText = ptrException;
        if CONTAINS(description,'Failed to open Queue') then
            set eText = 'Failed to open Output queue';
            END if;
        If CONTAINS(description,'WTX exception') then
            DECLARE CRLF CHARACTER CAST(X'0D0A09' AS CHAR CCSID 1208);
            set eText = REPLACE(eText, CRLF, '');
        END IF;   
       
          SET  flowname=SUBSTRING(Loc  FROM  0  FOR  POSITION('.'  IN Loc REPEAT-1));
         

        SET OutputRoot.XMLNSC = NULL;
       
       
        SET OutputRoot.XMLNSC.ERRORLOG.TimeStamp = CURRENT_TIMESTAMP;
        SET OutputRoot.XMLNSC.ERRORLOG.FlowName = flowname;
        SET OutputRoot.XMLNSC.ERRORLOG.ErrorCode = messageNumber;
        SET OutputRoot.XMLNSC.ERRORLOG.ErrorText = eText;
        SET OutputRoot.XMLNSC.ERRORLOG.ErrorDescription = description;
       
        --Uncomment the below code to Override Email Configuration
       
        -- Add recipient information to the EmailOutputHeader
        --SET OutputRoot.EmailOutputHeader.To = 'bhanu7miralce@gmail.com';
        --SET OutputRoot.EmailOutputHeader.Cc = 'kota.bhanumurty@gmail.com';
        --SET OutputRoot.EmailOutputHeader.Bcc = '<recipient email address>';

        -- Add sender information to EmailOutputHeader
        --SET OutputRoot.EmailOutputHeader.From = 'kota.bhanumurty@gmail.com';
        --SET OutputRoot.EmailOutputHeader."Reply-To" = '<reply email address>';

        -- Add subject to EmailOutputHeader
        --SET OutputRoot.EmailOutputHeader.Subject = 'Replaced by ESQL compute node.';

        -- Add SMTP server information to the LocalEnvironment
        --SET OutputLocalEnvironment.Destination.Email.SMTPServer ='';

        -- Create a new message body, which will be sent as the main text of the email.
        --SET OutputRoot.BLOB.BLOB = CAST('This is the new text for the body of the email.' AS BLOB CCSID 1208);
       
        RETURN TRUE;
    END;

    CREATE PROCEDURE CopyMessageHeaders() BEGIN
        DECLARE I INTEGER 1;
        DECLARE J INTEGER;
        SET J = CARDINALITY(InputRoot.*[]);
        WHILE I < J DO
            SET OutputRoot.*[I] = InputRoot.*[I];
            SET I = I + 1;
        END WHILE;
    END;

    CREATE PROCEDURE CopyEntireMessage() BEGIN
        SET OutputRoot = InputRoot;
    END;
END MODULE;

Exception from Try that will be caught in Catch:-

 After ParseExceptionMessage:-

 Make Below Configurations at

C:\Program Files\IBM\MQSI\9.0.0.0>mqsicreateconfigurableservice IB9NODE -c SMTP
-o MYEMAILSERVER
BIP8071I: Successful command completion.

C:\Program Files\IBM\MQSI\9.0.0.0>mqsichangeproperties IB9NODE -c SMTP -o MYEMAI
LSERVER -n serverName -v <<serverName>>
BIP8071I: Successful command completion.

C:\Program Files\IBM\MQSI\9.0.0.0>mqsisetdbparms IB9NODE -n smtp::EmailIdentity
-u <<EmailId>> -p <<Password>>
BIP8071I: Successful command completion.

C:\Program Files\IBM\MQSI\9.0.0.0>mqsichangeproperties IB9NODE -c SMTP -o MYEMAI
LSERVER -n securityIdentity -v EmailIdentity
BIP8071I: Successful command completion.