Messaging API: EventAccess service

[former_member692003]

Do you look for a way of realize more complex logic of your scenario SAP PI but you don’t want use SAP BPM?

If yes, then this blog post could be interesting for you.

Do you want to know how use Messaging API services and even those services that are not in official documentation?

If yes, then this blog is also  for you.

Do you think that knowing how the Message API works will bring you closer to understanding how SAP PI works?

If yes, then this blog post might be interesting for you

So, let's begin

In this blog I want to share an experience of using one of services of Messaging API - EventAccess service.

Using the EventAccess service, you can implement scenario logic, which is usually resolved with using the BPM Engine.

EventAccess service makes possible to "subscribe" to events occurring in the system.

These events can be very different: message errors, change of status of message, message archiving and others. All events are interfaces and inherit the interface EventCallBack. Each event is responsible for a specific event in the system.

Here is a complete list of events that can be in SAP PI:

• FatalMessageErrorEvent


• FinalMessageStatusEvent


• MessageLoggedEvent


• MessagePersistedEvent


• MessagesAchiveDeletedEvent


• MessagesArchivedEvent


• MessagesDeletedEvent


• MessageStatusEvent


• MessageStatusSyncEvent


• PerformanceMeasurementEvent


• RetriesExceededEvent


• SyncMessageCorrelationEvent


• SystemStatusEvent

If you want to handle event you have to:

1. Implement a handler class that inherits an interface of event (you can inherit several interfaces at once).


2. Register a handler with the EventAccess service.

Example

It is always better to explain with an example. So let’s come up with it. Let’s use the default SingleFlightBooking scenario. As a reminder, Agency sends asynchronous Booking Order message to Airline. Then Airline sends an asynchronous Process Order Confirmation message to Agency.

What happens if the Booking Order is not sent to Airline (for example due to an HTTP 500 error of FlightBookingOrderRequest_In interface)? In this case Airline will not receive the Booking Order and therefore will never send the Process Order Confirmation to Agency.

Let’s fix it.

Problem statement: if the message was not delivered to Airline (failed) the Agency should receive the Process Order Confirmation message with BookingStatus set to “F” (failed).

Many people would agree that in the PI world such a task must be accomplished using BPM or using the PI Alert subsystem. Both variants have their drawbacks. We will solve this task using a small custom application that uses the EventAccess service.

Prerequisites

For our application we need the following files:

• com.sap.aii.af.sdk.jar


• com.sap.aii.af.ms.ifc_api.jar


• com.sap.aii.adapter.xi.svc_api.jar

They can be copied from NW itself or taken from the MESSAGING*.SCA file distribution.

(Demo example Software Component FLIGHT_EXT with source codes, you can download from the this link. The same classes can be found in the FLIGHT_EXT component itself)

FlightEventHandler

In order to call our handler in case of an error it must inherit two interfaces: RetriesExceededEvent and FatalMessageErrorEvent.

The source code of the class is following:

What you should pay attention to: the onFatalMessageError and onRetriesExceeded are passed a MessageErrorData object that contain information about the error as well as the messageId of the failed message. MessageErrorData gives us everything we need at once to filter the events that we are interested in.

(To simplify the code in the example, the system names are defined by constants, but of course in a real application this should not be done).

package demo.flight.event;



import javax.naming.Context;

import javax.naming.InitialContext;

import javax.xml.parsers.SAXParser;

import javax.xml.parsers.SAXParserFactory;



import org.xml.sax.SAXException;

import org.xml.sax.helpers.DefaultHandler;



import com.sap.aii.adapter.xi.ms.XIMessage;

import com.sap.engine.interfaces.messaging.api.*;

import com.sap.engine.interfaces.messaging.api.event.*;

import com.sap.tc.logging.Location;



public class FlightEventHandler implements RetriesExceededEvent, FatalMessageErrorEvent {

	protected static final Location LOG = Location.getLocation(FlightEventHandler.class);

	

	public static final String AF_CONNECTION_FACTORY = "MessagingConnectionFactory";

	public static final String AF_CONNECTION_NAME = "AFW";

	public static final String XI_MESSAGE_FACTORY_NAME = "XI";



	

	private static final String SERVICE = "Airline";

	private static final String ACTION = "BookingOrderRequest_Out";

	private static final String ACTION_NS = "http://sap.com/xi/XI/Demo/Agency";

	private static final String ACTION_CONFIRM = "FlightBookingOrderConfirmation_Out";

	private static final String ACTION_CONFIRM_NS = "http://sap.com/xi/XI/Demo/Airline";

	

	

	private class FlightBookingOrderRequestHandler extends DefaultHandler {

		private String agencyID;

		private String orderNumber;

		

		private StringBuilder text;

		

		@Override

		public void startDocument() throws SAXException {

			text = new StringBuilder();

		}

		

		@Override

		public void characters(char[] ch, int start, int length) throws SAXException {

			text.append(ch, start, length);

		}

		

		@Override

		public void endElement(String uri, String localName, String qName) throws SAXException {

			if(localName.equals("AgencyID")){

				agencyID = text.toString().trim();

				text.setLength(0);

			}

			if(localName.equals("OrderNumber")){

				orderNumber = text.toString().trim();

				text.setLength(0);

			}

		}

		

		public String getAgencyID() {

			return agencyID;

		}

		

		public String getOrderNumber() {

			return orderNumber;

		}

	}

	

	@Override

	public void onFatalMessageError(MessageKey messageKey, MessageErrorData errorData) {

		String SIGNATURE = "onFatalMessageError(MessageKey,MessageErrorData)";

		LOG.entering(SIGNATURE);

		if(SERVICE.equals(errorData.getToService()) 

				&& ACTION.equals(errorData.getAction()) 

				&& ACTION_NS.equals(errorData.getActionNamespace())

				&& "SOAP_ADAPTER_PROCESSING_ERROR".equals(errorData.getErrorCode())){

			triggerErrorCallback(messageKey, errorData);

		}

		LOG.exiting(SIGNATURE);

	}



	@Override

	public void onRetriesExceeded(MessageKey messageKey, MessageErrorData errorData) {

		String SIGNATURE = "onRetriesExceeded(MessageKey,MessageErrorData)";

		LOG.entering(SIGNATURE);

		if(SERVICE.equals(errorData.getToService()) 

				&& ACTION.equals(errorData.getAction()) 

				&& ACTION_NS.equals(errorData.getActionNamespace())

				&& "SOAP_ADAPTER_PROCESSING_ERROR".equals(errorData.getErrorCode())){

			triggerErrorCallback(messageKey, errorData);

		}

		LOG.exiting(SIGNATURE);

	}



	

	private void triggerErrorCallback(MessageKey messageKey, MessageErrorData errorData){

		String SIGNATURE = "triggerErrorCallback(MessageKey,MessageErrorData)";

		LOG.entering(SIGNATURE, new Object[] {messageKey, errorData});

		

		try {

			Context ctx = new InitialContext();

			

			ConnectionFactory connectionFactory = (ConnectionFactory) ctx.lookup(AF_CONNECTION_FACTORY);

			Connection afConnection = connectionFactory.createConnection(AF_CONNECTION_NAME);

			afConnection.start();

			MessageFactory afMessageFactory = afConnection.createMessageFactory(XI_MESSAGE_FACTORY_NAME);

			

			Party senderParty = new Party(errorData.getToParty());

			Service senderService = new Service(errorData.getToService());

			Action action = new Action(ACTION_CONFIRM, ACTION_CONFIRM_NS);



			XIMessage xiMessage = (XIMessage) afMessageFactory.createMessage(senderParty, null, senderService, null, action);

			xiMessage.setInterface(ACTION_CONFIRM, ACTION_CONFIRM_NS);

			xiMessage.setMessageClass(MessageClass.APPLICATION_MESSAGE);

			XMLPayload xmlPayload = xiMessage.createXMLPayload();

			xmlPayload.setName("TechnicalError");

			xiMessage.setDeliverySemantics(DeliverySemantics.ExactlyOnce);

			

			SAXParserFactory factory = SAXParserFactory.newInstance();

			factory.setNamespaceAware(true);

			SAXParser saxParser = factory.newSAXParser();

			

			Message sourceMessage = APIAccessFactory.getAPIAccess().getMessageAccess().getMessage(messageKey); 

			FlightBookingOrderRequestHandler handler = new FlightBookingOrderRequestHandler();

			saxParser.parse(sourceMessage.getMainPayload().getInputStream(), handler);

			

			StringBuilder buf = new StringBuilder();

			buf.append("<ns0:FlightBookingOrderConfirmation xmlns:ns0=\"http://sap.com/xi/XI/Demo/Airline\">");

			buf.append("<AgencyData>");

			buf.append("<AgencyID>").append(handler.getAgencyID()).append("</AgencyID>");

			buf.append("<OrderNumber>").append(handler.getOrderNumber()).append("</OrderNumber>");

			buf.append("<ItemNumber/><OrderType/></AgencyData>");

			buf.append("<BookingStatus>F</BookingStatus>");

			buf.append("</ns0:FlightBookingOrderConfirmation>");

			xmlPayload.setContent(buf.toString().getBytes());

			xiMessage.setMainPayload(xmlPayload);

			afConnection.send(xiMessage);

		} catch (Exception e) {

			LOG.throwing(e);

			throw new RuntimeException("Failed send confirmation", e);

		}

		

		LOG.exiting(SIGNATURE);

	}

}

 

Registration handler

In order to call our handler it must be registered in the service.

Unfortunately SAP NW does not support EJB 3.1 where there is a singleton. So we use a trick with ServletContextListener. Let’s take advantage of its ability to call contextInitialized and contextDestroyed methods once at start and stop respectively.

The source code of the class which implement ServletContextListener interface is following:

What you should pay attention to: when registering EventCallback it is possible to set the handler call mode – synchronous or asynchronous. In our case, an asynchronous call is sufficient. The mode is set in the second parameter of registerEventCallback method.

package demo.flight.web;



import java.util.HashMap;

import java.util.HashSet;

import java.util.Iterator;



import javax.servlet.ServletContextEvent;

import javax.servlet.ServletContextListener;



import com.sap.engine.interfaces.messaging.api.APIAccessFactory;

import com.sap.engine.interfaces.messaging.api.event.*;

import com.sap.tc.logging.Location;



import demo.flight.event.FlightEventHandler;



public class InitServletContainer implements ServletContextListener{

	private static final Location LOG = Location.getLocation(InitServletContainer.class);

	

	private static  HashSet<EventCallback> events = new HashSet<EventCallback>();

	public static final String ATTRIBUTE_EVENT_LEVEL = "applicationLevel";

	public static final String ATTRIBUTE_EVENT_SYNC = "synchronous";

	

	@Override

	public void contextDestroyed(ServletContextEvent servletContextEvent) {

		String SIGNATURE = "contextDestroyed(ServletContextEvent)";

		LOG.entering(SIGNATURE, new Object[] {servletContextEvent});

		

		

		try {

			EventAccess eventAccess = APIAccessFactory.getAPIAccess().getEventAccess();

			Iterator<EventCallback> iterator = events.iterator();

			

			while(iterator.hasNext()){

				EventCallback event = iterator.next();

				eventAccess.unregisterEventCallback(event);

				LOG.debugT("Unregistered {0} event", new Object[] {event.getClass().getName()});

			}

		} catch (Exception e) {

			LOG.throwing(SIGNATURE, e);

			throw new RuntimeException(e);

		}

		LOG.exiting();

	}



	@Override

	public void contextInitialized(ServletContextEvent servletContextEvent) {

		String SIGNATURE = "contextInitialized(ServletContextEvent)";

		LOG.entering(SIGNATURE, new Object[] {servletContextEvent});

		

		

		try {

			EventAccess eventAccess = APIAccessFactory.getAPIAccess().getEventAccess();

			

			EventCallback event = new FlightEventHandler(); 

			HashMap<String, Boolean> map = new HashMap<String, Boolean>();

			map.put(ATTRIBUTE_EVENT_LEVEL, false);

			map.put(ATTRIBUTE_EVENT_SYNC, true);

			

			events.add(event);

			eventAccess.registerEventCallback(event, map);

			LOG.debugT("Registered {0} event", new Object[] {event.getClass().getName()});

		} catch (Exception e) {

			LOG.throwing(SIGNATURE, e);

			throw new RuntimeException(e);

		}

		LOG.exiting();

	}

}

to be called

Integration Directory

The problem statement assumes the presence of corresponding objects of Integration Builder (ICO, channels, etc). There is no need to make any special changes in the Integration Directory for FlightEventHandler to work (this is one of the advantages of our approach).

Testing

Send test message Booking Order from Agency to Airline. In order the message can receive the HTTP 500 error it is necessary to redirect receiver channel Airline to mock service which answers with HTTP 500 error.

We see:

1. The sent test message from Agency to Airline failed with error.


2. SAP PI created and sent the confirmation message to Agency where BookingStatus set to “F”.

I used EventCallBack events for the following tasks:

• to catch error message


• to send an acknowledgment of receipt of the message by the receiver system to the sender system

It would be interesting to know how EventAccess could be useful for you?