If X do Y – how to listen to (internal) SAP CPI events

Since the last excursion into the depths of the internal SAP CPI functions was well received, today we are launching another excursion into the framework that underlies SAP CPI.

Those of you who have worked not only with the WebIDE but also with the SAP CPI eclipse plugin may well know the “Task View”. This small view in Eclipse shows you when a deployment is running or completed on the SAP CPI, but also when content is synchronized between the nodes. Today we want to learn, how we can attach our own code/interfaces to this events.

Attaching to those events allows us a bunch of new scenarios. You can not imagine any? How about:

• “Whenever an IFlow is deployed, send an e-mail to the maintenance team.”
• “Make a backup of the current IFlows whenever a deployment happens.”
• …

I think these are pretty cool scenarios and there are certainly more use cases. Before we start, let’s take a quick look at the contents of this blog post…

Table of Contents

• Introduction to OSGi events
• A practical use case
• Building your first OSGi EventHandler (Integraton Flow)
• Outlook, ideas and list of events
• Feedback and Discussion

Introduction to OSGi events

To understand what we want to do, it is necessary to have a rough understanding of how SAP CPI works. Let me try to give you a short introduction in simple words. When you deploy IntegrationFlows they are “translated” into Java program code, which is heavily based on the Apache Camel framework. (A framework that helps to build integration applications.) Since CPI is modularized those IFlow java programs aren’t just simply executed, but ran as OSGi bundles.

OSGi itself is a framework which (in simple words) allows modularization of software. Each IFlow will be handled as so called OSGi bundle. The OSGi framework than allows to start, stop, deploy, run those bundles. Not only IFlows, but also other core components of the SAP CPI are built/handled as OSGi bundles. And here we come to a sticking point: How do the SAP CPI tool bundles communicate with the IFlow bundles? How does the WebIDE / monitor bundle know that an IFlow (bundle) has been successfully deployed?

Due it’s modularized nature OSGi comes with an Event-Service which allows bundles to send and receive messages. Let’s have a look at the following graphic. (Please excuse the layout – I’m a technician, not a designer…)

Bundles can act as so called event publishers. An event publisher can send events to the EventAdmin-service whereby each event has two main characteristics:

1. Event topic: Each event has a topic in the form of /general/topic/more/detailed
2. Properties: Each event can have 0..n properties (=additional information/payload)

The events then will be “shared” by the EventAdmin service. On the other side you will find the event handlers (based on OSGi’s EventHandler-class). Event handler can be registered on OSGi’s context and listen for messages of topic, defined in the event handlers properties. Each an event with a matching topic flows by, the event handler’s “handleEvent” method is triggered. Inside the handleEvent method the event handler gets hold of the event including it’s properties and can to further actions/processing.

Please note the following things regarding event handlers:

• Multiple event handlers can listen to the same topic (and all will get hold of the event)
• An single event handler can listen to multiple topics (1..n)
• An event handler can use wildcards for the topic definition. E.g. /com/source/* would match events with the topic /com/source/event_a as also /com/source/event_b

A practical use case

Let’s have a view onto the following story. Mike is just one deployment away from his weekend, but then the deployment fails…

(Click the image to enlarge.)

Somehow the endpoint Mike should use is already in use. But who did it? Which IFlow uses the endpoint Mike wanted to use in his IFlow and even if he finds the IFlow – is he allowed to undeploy it?

Let’s imagine there would be an automatic e-mail, everytime someone deploys an artifact. Would mike be able to go to the weekend punctually?

(Click the image to enlarge.)

Seems like an deployment info mail could be a good thing. So let’s build it! The goal is to automatically send an email containing information about who/when/what was deployed on each deployment.

Building your first OSGi EventHandler (Integration Flow)

The Integration Flow we want to build during this article, shall fulfill the following three purposes:

1. Register a custom OSGi event handler
2. (Optionally) un-register an OSGi event handler
3. Handle events, sent in by our custom event handler, within the IFlow

The following guide will take care of this three functional requirements in the order as listed above. At first create a new IFlow with the elements shown as in the screenshot below. (The following detailed described refers to the numbers in the screenshot.)

(1) Start with a timer element. Set it to run X times a day. The IFlow will then check several times a day if our custom handler is still registered. (We use this approach, because it might be that our custom event handler will be de-registered due to a reboot or similiar situations. By starting with a timer, we can ensure, that our custom event handler will be re-registered in such a case.

(2) Place a content modifier as next step. Configure the following properties as shown in the screenshot below.

• The property eventListenType should be set to the name of the OSGi event we want to listen to. For CPI artifact deployment the event is named “com/sap/it/nm/events/content/DEPLOYED“.
• The property endpointBase should contain the base url of your CPI instance (including http adapter base path). Eg.: https://<tenant-id>-iflmap.hcisbp.<region>.hana.ondemand.com/http
• The property endpointSlug should contain the endpoint path, we will configured later for our event handler endpoint. Enter the same value here as you will use later in the HTTP sender channel (or use externalize parameter feature to re-use the pattern). E.g.: /pr009/deploymenthandler
• The property deleteEventhandler should be empty(!). We will use it as switch. If it is filled, the IFlow will run in “de-register event handler”-mode
• The property credentialName should be filled with the name/id of a SAP CPI security material which contains user credentials to access the SAP CPI. (=S-User and password)
• The property eventHandlerId should be filled with a custom id of your favour. This will be used just as “tag” to be able to identify your event handler later in de-registering mode. (If you like to create multiple eventhandler IFlows, take care that each has a different eventHandlerId!)

(3) This route should be used, if in “event handler registration”-mode. As explained in (2) we decide via the property deleteEventhandler which mode the IFlow is in.

(4) In this Groovy script step, we want to check all registered event handlers and proof if our custom event handler is registered. To recognize our handler we use the eventHandlerId we defined in the content modifier in (2). Since the script is a little bit longer, I added a lot of sourcecode comments for you. (If anything is unclear, write a comment!)

import com.sap.gateway.ip.core.customdev.util.Message;
import java.util.HashMap;
import groovy.json.JsonOutput
import org.osgi.framework.*
import org.osgi.service.event.*

def Message processData(Message message) {
    
    //Read custom event handler id from message properties
    def eventHandlerId = message.getProperty("eventHandlerId")
    
    //Retrieve a list of EventHandler information
    def res = listOSGi()
    
    //Check if own EventHandler is registered (by counting matching eventhandlers)
	def numOwnEvents = res.findAll{ 
	    it.props.any { 
	        prop -> prop.getKey() == "custom.id" 
	             && prop.getValue() == eventHandlerId 
	    }
	}.size()
	
	//Store custom event handler, so that we can use it in a route-block later
	message.setProperty("numOwnEvents", numOwnEvents)
	
    return message;
}

//This function returns a list of registered EventHandler objects
private listOSGi(){

	//Get an OSGi general bundle context
	def bundleCtx = FrameworkUtil.getBundle(Class.forName("com.sap.gateway.ip.core.customdev.util.Message")).getBundleContext()
	
	//Get a complete list of service references of type (OSGi) EventHandler
	ServiceReference[] srs = bundleCtx.getServiceReferences(EventHandler.class.getName(), null)
	
	//Create data object to store event handler information
	data = []
	
	//Loop through all results and read contents
	srs.each { ref -> 
	    //Create map of all properties of current eventhandler
		def props = [:]
		ref.getPropertyKeys().each { propKey -> 
			props <<  ["$propKey":ref.getProperty(propKey)]
		}
		//Add eventhandler info (incl. properties map) to data array
		data << [
			name:ref.getBundle().getSymbolicName(),
			propCount:props.size(),
			props:props
		]
	}
	return data
}

In a nutshell: The script calls our listOSGi-function which returns a list of registered OSGi event handlers. Then we check if the list one or more references to our custom handler and store the number into a property called “numOwnEvents”. Now that we know if our custom event handler is already registered we can proceed in our flow.

(5) This route shall be used, if our custom event handler is already registered. In this case, the message flow should end. (That’s why the route points to a “message end”-event.) The rule used in the route looks as follows:

(6) This route shall be used, if our custom handler wasn’t found in (4). Since it’s “true”-case is the opposite of the route (5), we can just set it as “Default Route”.

(7) This script step contains the most “complex” logic of our IFlow. It is the script which defines the custom EventHandler and registers it at the OSGi context. Since it is even longer than the last script, I added again a lot of code comments. (If something is not clear, write a comment.)

import com.sap.gateway.ip.core.customdev.util.Message
import groovy.json.JsonOutput
import org.apache.camel.*
import org.osgi.framework.*
import org.osgi.service.event.*
import com.sap.it.api.ITApi
import com.sap.it.api.ITApiFactory
import com.sap.it.api.securestore.*

def Message processData(Message message) {
    
    //Get message properties, which were set in ContentModifier (1)
    def eventHandlerId = message.getProperty("eventHandlerId")
    def credentialName = message.getProperty("credentialName")
    def endpointBase = message.getProperty("endpointBase")
    def endpointSlug = message.getProperty("endpointSlug")
    def eventListenType = message.getProperty("eventListenType")
    
    //Build endpoint url, of the endpoint of this IFlow
    def targetUrl = "${endpointBase}${endpointSlug}"
    
    //Register our custom IFlow event
    def res = registerOSGiEvent(eventHandlerId, credentialName, targetUrl, eventListenType)
    
    return message
}

/***********************************************
 * This function helps registering OSGi events *
 ***********************************************/
private registerOSGiEvent(def eventHandlerId, def credentialName, def targetUrl, def eventListenType){
	//Get general bundle context
	def bundleCtx = FrameworkUtil.getBundle(Class.forName("com.sap.gateway.ip.core.customdev.util.Message")).getBundleContext()
	
	//Define the topics we like to listen to
	def topics = [
        eventListenType
    ]
    //Configure our event listener
    def props = new Hashtable();
    props.put(EventConstants.EVENT_TOPIC, topics)
    props.put("custom.id", eventHandlerId) 
    props.put("custom.targeturl", targetUrl)
    
    //Get credentials and build Basic Authentication string
    def creds = getUserCreds(credentialName)
    def bAuth = creds.basicAuth
    
    //Register custom EventHandler as service and pass properties, credentials 
    //and endpoint url to the event handler
    bundleCtx.registerService(EventHandler.class.getName(), new DeployEventHandler(bAuth, targetUrl), props)
    return [successful:true]
}


/***********************************************************
 * This helper-function retrieves credential data via name *
 ***********************************************************/
private getUserCreds(def credentialName){
    //Get credential service instance
	def service = ITApiFactory.getApi(SecureStoreService.class, null)
	//Read security material/credential
	def credential = service.getUserCredential(credentialName)
	//Return credential object incl. BasicAuth string
    return [
    	username:credential.getUsername(),
    	password:new String(credential.getPassword()),
    	basicAuth:"Basic ${"${credential.getUsername()}:${new String(credential.getPassword())}".bytes.encodeBase64().toString()}"
    ]
}

/**************************************************************
 * This is the custom eventhandler class, we want to register *
 **************************************************************/
public class DeployEventHandler implements EventHandler
{
    private String _basicAuth
    private String _targetUrl
    
    //Take endpoint url and auth string in constructor
    public DeployEventHandler(String basicAuth, String targetUrl){
        _basicAuth = basicAuth
        _targetUrl = targetUrl
    }
    
    //This function will be called everytime, when an 
    //event with a matching topic passes by. Everything which
    //should happen at an event, must be implemented here.
    public void handleEvent(Event event)
    {
        //The complete code is called as "async" Runnable in a different thread,
        //because OSGi has a default timeout of 5000ms for events. If an event-
        //handler takes more time, it will be blacklisted. By use of Runnable, 
        //we can bypass this limit.
    	Runnable runnable = {
    	    //Build event information
    		def evntMsg = [topic:event.getTopic(),name:"From within demo IFlow"]
    		//Build properties table
	    	evntMsg.properties = []
			event.getPropertyNames().each { propKey -> 
				evntMsg.properties << ["$propKey":event.getProperty(propKey)]
			}
            //Send event information to our own IFlow via HTTP post.
            //Thus we can handle the event within our IFlow. :-)
			def post = new URL(this._targetUrl).openConnection()
			post.setRequestMethod("POST")
			post.setDoOutput(true)
			post.setRequestProperty("Authorization", this._basicAuth)
			post.setRequestProperty("Content-Type", "application/json")
			post.getOutputStream().write(JsonOutput.toJson(evntMsg).getBytes("UTF-8"))
			def postRC = post.getResponseCode()
    	}
    	//Call the Runnable
	    def thread = new Thread(runnable);
	    thread.start();
    }
}

(8) & (9) Right after the event handler registration we use a external call/send module to send out an e-mail to inform me that the event handler was successfully registrated. Below you can find the mail channel configuration for a GMail server. (Mail text, subject, etc.were set “manually”.)

(10) As last step on this route, we end the message flow with an “Escalation End” event. (You could also use an EndMessage event, but I wanted to be able to filter for “escalated” messages in the monitoring to quickly find the one, were the registration was done.

---

Interim state: Right now you have managed to implemented the first of our three requirements. (1: Register event handler, 2: De-register event handler, 3: Handle event in IFlow). In the next steps (11-13) we will implement the functionalty, which allows us to unregister the custom eventhandler, which gets registered when the code from (7) is executed.

---

(11) This route shall be set as “default” route, because we have a “either … or” decision here. By setting it to “default route” it will be run every time the other condition isn’t true. Which means if you set the property deleteEventhandler to “X”, you will end up on this route. (Because the other route checked for deleteEventhandler = ”.)

(12) Since we are on the “delete event handler”-route, the next script will do exactly this. It will search for event handlers which match our event handler id (see (2)) and un-register them.
As for the other script I also let this speak for itself (by its source code comments).

import com.sap.gateway.ip.core.customdev.util.Message
import java.util.HashMap
import org.apache.camel.*
import org.osgi.framework.*
import org.osgi.service.event.*

def Message processData(Message message) {
    
    //Read the name of the event handler from properties
    def eventHandlerId = message.getProperty("eventHandlerId")
    //Call unregister function with custom eventhandler id
    def res = unregisterOSGiEvent(eventHandlerId)
   
    return message;
}

/****************************************************************
 * This function de-registers an event handler by its custom id *
 ****************************************************************/
private unregisterOSGiEvent(def eventHandlerId){
	//Get general bundle context
	def bundleCtx = FrameworkUtil.getBundle(Class.forName("com.sap.gateway.ip.core.customdev.util.Message")).getBundleContext()
	//Get all service references that match our eventhandler id
	ServiceReference[] srs = bundleCtx.getServiceReferences(EventHandler.class.getName(), "(custom.id=${eventHandlerId})")
	//For each service reference found...
	srs.each { sRef -> 
	    //...get registration and unregister it!
		sRef.getRegistration().unregister()
	}
    return [successful:true, count:srs?.size()]
}

(13) After deregistration we send out an e-mail so that you have a reminder in your inbox, that you decided to un-register…

---

Interim state: Great – no you’re unable to register and un-register your custom event handler. But wait – where do we handle this events? If you step back to the bottom of the registration script in (7), you can see that our event handler just passes the event to another (SAP CPI) url. Right now it doesn’t exist. So let’s build it in the next steps to fulfill our third requirement (“Handle events, sent in by our custom event handler, within the IFlow“).

---

As we have already gone through all 13 steps of the last IFlow overview graphic, it is time for another graphic/part. The details of the individual points will follow again afterwards.

As first, add a second Integration Process to your IFlow. Then add the sender element and connect it via HTTPS-channel (14). Next add a Groovy Script (15), connect it with the “End Message” event and connect the “End Message” event itself via a Mail receiver (16) with the receiver system. Now let’s have a look into the details.

(14) This sender channel is the channel which will receive the event message which will be captured by the custom event handler we registered beforehand. Thus the senders endpoint must match the one configured in step (2) while setting the endpointSlug-parameter.

(15) In this step we use a Groovy script, to parse the SAP CPI deployment event and build a nice HTML-formatted e-mail body.

import com.sap.gateway.ip.core.customdev.util.Message
import java.util.HashMap
import groovy.json.JsonSlurper

def Message processData(Message message) {
    
    //Get captured SAP CPI event json from body
    def body = message.getBody(java.lang.String) as String
    
    //Get message log factory and save the captured event
    //into the message log, thus we can inspect it in the 
    //regular monitoring.
    def messageLog = messageLogFactory.getMessageLog(message);
    if(messageLog != null){
        messageLog.addAttachmentAsString("DeploymentEvent", body, "application/json");
     }
    
    //Parse the event json via JSON slurper
   	def eventJson = new JsonSlurper().parseText(body)
   	//Get a shorthand to the descriptors property, because nearly all information 
   	//come from this note.
    def descriptors = eventJson.properties.find{it.'artifact.descriptors' != null}.'artifact.descriptors'[0]
	
	//Create a HTML-formatted property list with information about the deployment
    def propertyList = createListEntry("Artifact Name", descriptors.name)
    propertyList += createListEntry("Artifact Id", descriptors.symbolicName)
    propertyList += createListEntry("Deployed", "On ${descriptors.deployedOn} by ${descriptors.deployedBy}")
    def ipAddress = descriptors.tags.find{it -> it.name == "initiating.source"}.value
    propertyList += createListEntry("Deployed from", "${ipAddress} (${tryResolveCity(ipAddress)})")
    propertyList += createListEntry("Last modfied", "On ${formatTimestamp(descriptors.lastModifiedTime)} by ${descriptors.lastModifiedByUser != -1 ? descriptors.lastModifiedByUser : ""}")
    propertyList += createListEntry("Init. created", "On ${formatTimestamp(descriptors.createdTime)} by ${descriptors.createdBy}")
    
	//Create HTML mail body and insert our properties html
   	def mailBody = """
Dear CPI-Admin,<br/>
<br/>
There was a deployment on <i>${new URI(message.getHeaders().get('CamelHttpUrl')).getHost()}</i> with the following properties:<br/>
<ul>
    ${propertyList}
</ul>
<br/>
Regards,<br/>
CPI-IFlow
   	"""
	//Set mail body as message body
   	message.setBody(mailBody)
   	message.setHeader("Content-Type", "text/html")
    return message
}

/*****************************************************
* Small helper function to create html <li>-elements *
******************************************************/
def createListEntry(def key, def value){
	return "<li><b>${key}</b>: ${value}</li>"	
}

/********************************************
* Small helper function to format timestamp *
*********************************************/
def formatTimestamp(def ts){
	return new Date(ts).format("yyyy-MM-dd'T'HH:mm:ssZ")
}

/***************************************************
* This function can resolve IP address to location *
****************************************************/
def tryResolveCity(def ip){
	try {
		def ipJson = "http://ip-api.com/json/${ip}".toURL().getText([requestProperties:["Accept":"application/json"]])
		def jsonRoot = new JsonSlurper().parseText(ipJson)
		def location = jsonRoot.country
		if (jsonRoot.city != null && jsonRoot.city != ""){
			location += ", ${jsonRoot.city}"
		}
		if (jsonRoot.org != null && jsonRoot.org != ""){
			location += " @ ${jsonRoot.org}"
		}
		return location
	} catch (Exception ex){
		return "unknown location"
	}
}

(16) In this last step we send out the mail to ourselves (or whomever). In difference to the other two mail channels we take the body from the message body now. In addition we have to switch the Body Mime-Type to “Text/HTML” to get the mail rendered the right way.

That’s it – we’re done! Just setup the parameters in (2) to your needs (especially endpoint and eventhandler id), deploy the IFlow and look forward to the info emails you get with each deployment.

Outlook, ideas and list of events

Now that we know how to create our own event handlers, there are no limits to creativity. To make starting your own projects a little easier, I would like to give you two things: a list of ideas and a list of events to listen to.

(Event-)Topics you can listen to

Each event has a topic, as you’ve learned. In our example we listened to the com/sap/it/nm/events/content/DEPLOYED event. But there’s more you can listen to. Below you find a list of events, I collected by listing all registered event handlers in our CPI system and check out to which topics they are listening to.

• com/sap/esb/camel/core/event/polling/strategy/notification/*
• com/sap/esb/camel/core/event/polling/strategy/query
• com/sap/esb/security/keyring/*
• com/sap/esb/security/keyring/public/query
• com/sap/esb/security/keyring/secret/query
• com/sap/esb/security/keystore/notification/*
• com/sap/esb/security/keystore/query
• com/sap/it/config/cache/INVALIDATE
• com/sap/it/nm/audit/LOGGED/*
• com/sap/it/nm/component/CHANGE
• com/sap/it/nm/content/ACTIVATE
• com/sap/it/nm/content/DEACTIVATE
• com/sap/it/nm/events/content/DEPLOYED
• com/sap/it/nm/events/content/UNDEPLOYED
• com/sap/it/nm/events/task/deploy/COMPLETED
• com/sap/it/nm/keyring/*
• com/sap/it/nm/keystore/*
• com/sap/it/nm/lm/tenant/REMOVE
• com/sap/it/nm/lm/tenant/UPDATE
• com/sap/it/nm/node/lm/DEPLOYED
• com/sap/it/nm/node/lm/LAUNCH
• com/sap/it/nm/node/password/*
• com/sap/it/nm/node/uri/SYNC
• com/sap/it/nm/ssh/CHANGE
• com/sap/it/pd/CACHE/INVALIDATE
• com/sap/it/pd/CHANGE
• org/apache/camel/*
• org/apache/camel/context/*

Note: Don’t take this list as complete. As I mentioned, I compiled it by taking all topics I could find listening event handlers for. There could be more events send inside your SAP CPI tenant, for which just no event handler is registered. To find those you had to register an event handler for the topic * (everything) and analyize all events you received.

Ideas for use cases

You want to build something cool with the things you learned in this article? How about…

• On deployment download/read the IFlow object, unzip it and commit/push it to Git(hub). Thus you could “auto archive” each productive version.
• On failed deployment sent out a mail to the SAP user who deploy the IFlow
• On Keystore/Keyring changes send out an info to the manager
• …

Feedback and Discussion

As in my other “deep dives”, too, this was again a fun journey for me. I learned and lot and hopefully could share it in a way that you can take some new information from. If something is unclear or if you think that parts of the article are not clear enough, let me know. Then I’ll see what I can do.

Last but not least, I would like to know if you worked with the EventAdmin/EventHandler before or if everything in this article was new for you. Also I would love to see the results, if you decide to build your own event listeners. (So if you build one, share your ideas in another SAP blog or in the comments.) Until then – happy coding!