Working with Event Sinks

Event sinks contain information about transport endpoints provided by other systems. Events can be published to these endpoints. Endpoint transport and authentication URLs and credentials are captured in event sinks.

Adding an event sink

Follow the steps below to add an event sink.

1. Open the ESB Management Console. On the Configure tab, click Event Sinks to open the Event Sinks Configuration page.
2. Click Add Event Sink to open the Add Event Sink page.
   

   

   Enter values for the following parameters.

   Parameter Name	Description
   Name	The name of the event sink.
   Username	The user name of the data-bridge endpoint.
   Password	The password of the data-bridge endpoint.
   Receiver URL	The URL of the data-bridge endpoint. For example, tcp://10.200.3.218:7611 Multiple endpoints can be specified for load balancing and multiple receiver modes, by using the correct URL format.
   Authenticator URL	The secure data-bridge endpoint URL that should be used for authentication. For example, ssl://10.200.3.218:7711 The URL specified here should match the specified Receiver URL parameter.

3. Click Save.

Sample

Let's look at a sample where WSO2 ESB uses event sinks to publish mediation data to WSO2 Data Analytics Server (WSO2 DAS).

Setting up WSO2 Data Analytics Server

1. Download and install WSO2 DAS. For instructions on how to download and install WSO2 DAS, see Getting Started with WSO2 DAS. 

   The unzipped WSO2 DAS distribution folder will be referred to as <DAS_HOME> throughout the documentation.

   It is not possible to start multiple WSO2 products with their default configurations simultaneously in the same environment. Since all WSO2 products use the same port in their default configuration, there will be port conflicts. Therefore, to avoid port conflicts, apply a port offset in the  <DAS_HOME>/repository/conf/carbon.xml  file by changing the offset value to 1 as follows:

   <Ports>
           <!-- Ports offset. This entry will set the value of the ports defined below to
            the define value + Offset.
            e.g. Offset=2 and HTTPS port=9443 will set the effective HTTPS port to 9445
            -->
           <Offset>1</Offset>

2. Open a command prompt and go to the <DAS_HOME>/bin directory.
3. Start WSO2 DAS by executing sh wso2server.sh (on Linux/OS X) or wso2server.bat (on Windows).

Setting up WSO2 ESB

1. If you have not already done so, see Getting Started with WSO2 ESB for details on installing and running WSO2 ESB.
2. Log in to the management console at https://localhost:9443/carbon/ using admin for the username and password.

3. Create an event sink in WSO2 ESB by navigating to Configure ->Event Sinks in the management console. Click on Add Event Sink.
   
   Fill in the details as in the following table:

   Field	Value
   Name	das_event_sink
   Username	admin
   Password	admin
   Receiver URL	tcp://localhost:7612
   Authenticator URL	ssl://localhost:7712

   In WSO2 DAS the default thrift port and the authentication port are 7611 and 7711 respectively. Due to the port offset you configured the new ports are 7612 and 7712 as defined above.

4. Create a REST API by navigating to Main ->APIs in the management console and click on Add API. Fill in the details as in the table below:

   Field	Value
   API Name	StockQuoteAPI
   context	/stockquote

   Click on Switch to source view.

   

   In source view, add the following API configuration where the publish mediator uses the created event sink das_event_sink to publish to an event stream named stockquote_stream_1.0.0:

   <api xmlns="http://ws.apache.org/ns/synapse" name="StockQuoteAPI" context="/stockquote">
      <resource methods="GET" uri-template="/view/{symbol}">
          <inSequence>
              <payloadFactory media-type="xml">
                  <format>
                      <m0:getQuote xmlns:m0="http://services.samples">
                          <m0:request>
                              <m0:symbol>$1</m0:symbol>
                          </m0:request>
                      </m0:getQuote>
                  </format>
                  <args>
                      <arg evaluator="xml" expression="get-property('uri.var.symbol')" />
                  </args>
              </payloadFactory>
              <header name="Action" value="urn:getQuote" />
              <publishEvent>
                  <eventSink>das_event_sink</eventSink>
                  <streamName>stockquote_stream</streamName>
                  <streamVersion>1.0.0</streamVersion>
                  <attributes>
                      <meta>
                          <attribute xmlns:ns="http://org.apache.synapse/xsd" name="http_method" type="STRING" defaultValue="" expression="get-property('axis2', 'HTTP_METHOD')" />
                          <attribute xmlns:ns="http://org.apache.synapse/xsd" name="destination" type="STRING" defaultValue="" expression="get-property('To')" />
                      </meta>
                      <correlation>
                          <attribute xmlns:ns="http://org.apache.synapse/xsd" name="date" type="STRING" defaultValue="" expression="get-property('SYSTEM_DATE')" />
                      </correlation>
                      <payload>
                          <attribute xmlns:m0="http://services.samples" name="symbol" type="STRING" defaultValue="" expression="$body/m0:getQuote/m0:request/m0:symbol" />
                          <attribute name="request" type="STRING" defaultValue="" value="getSimpleQuote" />
                      </payload>
                      <arbitrary />
                  </attributes>
              </publishEvent>
              <send>
                  <endpoint>
                      <address uri="http://localhost:9000/services/SimpleStockQuoteService" format="soap12"/>
                  </endpoint>
              </send>
          </inSequence>
          <outSequence>
              <send />
          </outSequence>
      </resource>
   </api>

5. Deploy the back-end service  SimpleStockQuoteService. For instructions on deploying sample back-end services, see Deploying sample back-end services.

6. Start the Axis2 server. For instructions on starting the Axis2 server, see Starting the Axis2 server.
   

7. Open a command line terminal and enter the following request to invoke the API you defined above: 

   curl -v http://localhost:8280/stockquote/view/WSO2

   You will see the response from the SimpleStockQuoteService. 

Exploring published data in WSO2 DAS

1. Log in to the management console of WSO2 DAS server at https://localhost:9444/carbon/ using admin for the username and password.
2. You can view the event stream that was published from ESB by navigating to Main->Streams in the management console. 
   
3. To persist the event stream, click on Edit next to the event stream and then click on Next (Persist Event) at the bottom of the page.
4. Select the check box Persist Event Stream and then select the attributes that need to be persisted by selecting the relevant check boxes. In this case, select all attributes by selecting the check boxes Persist Attribute under Meta Data Attributes, Correlation Data Attributes and Payload Data Attributes. See WSO2 DAS documentation for details on persisting data.
   
   

   Click Save Event Stream.

5. You need to add an Event Receiver in WSO2 DAS to receive events. In WSO2 DAS, navigate to Main->Receivers and click on Add Event Receiver and fill in the following information:

   Field	Value
   Event Receiver Name	ESB_RECEIVER
   Input Event Adapter Type	Select wso2event

   Click Add Event Receiver.

   

6. You are now done with persisting the event stream definition and adding a receiver for the event stream. Once again invoke the API in ESB to publish events to the defined stream. In WSO2 DAS navigate to Main->Data Explorer and select the table STOCKQUOTE_STREAM from the drop down list and click on Search. You can then view the data published from ESB in the table as follows: