Using JBoss ESB and JBPM for Implementing VMS Solutions

Vertical Market Solutions (VMS) is an organization within NAVTEQ that provides custom solutions for its clients, including mobility portals and connected navigation systems. These solutions combine both NAVTEQ-provided services and third-party services to deliver composite services and content in the customer-specific forms, including Web services, WAP, portals, etc.

The challenge for VMS in responding to numerous sales opportunities as well as anticipating future customer requests requires the development of composable services that can be integrated quickly and easily into complete customer solutions. A middleware that enables the service integration is essential to delivering such systems.

In this article I will discuss how JBoss middleware platform, specifically JBoss ESB and jBPM (JBoss Business Process Management) can be used for building such systems.

Overall solution approach

Before delving into details of JBoss middleware products I will quickly describe here our basic approach for building VMS solutions.

Currently, NAVTEQ provides all of the functionality required for such solutions, including powerful location-based services such as Geocoding and Routing, traffic-related services such as information about accidents, traffic jam factors calculation and traffic alerts for a given route, and general ecommerce functionality including purchasing, authorization, and entitlements management.

The issues with directly exposing these services to the consumers are multifold:

• Every service serves a particular purpose, while NAVTEQ customers want a composite service. For example, a customer might require traffic and event alerts around his route. This can be implemented in several steps – geocode start, end and via points, calculate route, get events information around the route, get traffic information for a route. This sequence of steps constitutes a new composite service that should be created based on users requirements.
• Existing functionality was growing organically and as a result all of these services are using different data models/definitions, which make their usage and orchestration complex.
• Different customers might require content/services delivery via different mechanisms. Some might want the ability to directly access services using either SOAP or REST, some might want html/javascript fragments delivery to their web pages/portals, some might require WAP interfaces to services, etc.
• Existing implementations have their own security, user management, monitoring and management architecture/implementation.

As a result, for our implementation, we decided to use overall architecture, presented at Figure 1.

Figure 1 Overall Solution Architecture

In the center of this architecture is VMS Platform, implementing services required by specific customers by orchestrating existing back-end functionality. The platform consists of three main layers:

• Back end adapters – a layer responsible for accessing existing back-end (or/and 3^rd party) functionality and bringing the result of execution (in the form of standard domain model) into a platform.
• Service implementation layer - responsible for orchestration of existing services (accessible through back-end adaptors) into services required by customers. Additionally, some custom functionality that does not currently exist can be implemented in this layer.
• Consumer adapters – a layer responsible for delivering results of service execution to different types of clients over different protocols/transports.

In the rest of the article I will discuss how this architecture can be implemented using JBoss middleware platform.

JBoss ESB

JBoss ESB is an open source implementation of the Enterprise Service bus, based on The RosettaNet ESB and supporting creation, deployment and integration of services.

Architecturally, everything in the Jboss ESB is a service. These services are not Web Services – they are ESB services, which can be exposed over variety of transports. ESB services are single method (doWork) services that can be described using the following interface (shared by all services):

Message output = service.doWork(Message input)

An ESB message is somewhat modeled after a SOAP message and consists of several parts, including header, body, fault, attachments, etc. Every part can contain a collection (map) of serializable java objects that can be accessed by name. This means that JBoss ESB messages are not strongly typed and care should be used when implementing access to the message (type conversion).

A service definition should have the following format:

• Service Category/Name – Jboss ESB is keeping an internal registry of services (services self-register during their deployment). JBoss ESB internally decides on the optimal mechanism for accessing the services.
• Components of the input message – including their names (in the body of incoming message) and type.
• Components of the output message – including their names (in the body of outgoing message) and type.

Unlike traditional ESB implementations which are build using a set of interceptors, allowing the injection of additional processing in the service invocation pipeline, JBoss ESB implements a service as an explicit pipeline that can contain arbitrary set of actions (Figure 2), each of which implements either (potentially partial) service business functionality or infrastructure support, for example storing service message in the message store.

Figure 2 JBoss ESB Service

Unlike the traditional interceptor–based technique allowing for a better separation between “business” (service implementation) and infrastructure (interceptors) concern, JBoss ESB lumps both together in the execution pipeline. This is not without benefit, however, as the inclusion of all actions in a single service pipeline can simplify understanding of the overall service processing.

A service action in JBoss ESB is a java class, implementing the org.jboss.soa.esb.actions.ActionLifecycle interface. A service implementer typically decomposes service functionality in a set of actions and then implements each one of them. Such organization of service implementation provides JBoss ESB developers with two levels of reuse – services themselves and actions that can be reused across multiple services. A rich library of reusable actions is also provided by JBoss. This library includes the following types of actions¹:

• Transformers & Converters
• JBPM support
• Scripting
• Routing
• Web Service support
• Misc

A service request can be delivered by a variety of transports. A service definition includes references to the set of transports/endpoints which that service listens on (this endpoint has to be unique for a given service). Service definition also allows for configuration of a service throughput through specifying (on a transport basis) the number of threads dedicated to the processing of the service request.

Both set and sequence of service actions invocations and the endpoint the service is listening on are specified in jboss-esb.xml file. This allows to configure a given service based on a set of actions (either provided by JBoss. or specific for a given company or service). Such support for external XML configuration of actions further improves potential for actions reuse.

Service invocations in JBoss ESB

As we have mentioned above, everything in JBoss ESB is a service with Asynchronous² Invocation being the default pattern. Synchronous invocations are supported as well and are implemented by using correlations. Synchronous vs asynchronous invocations of ESB services can be controlled in two ways:

• The service implementer can define the service behavior using the message exchange pattern (MEP) parameter. Default MEP is Request-Response, meaning that service is capable of returning a response. If MEP is set to one-way, a service cannot and will not return a reply
• The service consumer can define whether he needs a reply from the service. If requested, a service with the request-response MEP will return a reply³.

To simplify service invocation, Jboss ESB provides the org.jboss.soa.esb.client.ServiceInvoker class, which supports a very simple interface for service invocation, based on a category/service name.

Local service invocations in JBoss ESB

In addition to remote transports, JBoss provides very fast local (within the same JVM) invocation mechanism that provides service invocation through an In-Memory queue. If the org.jboss.soa.esb.client.ServiceInvoker class is used for the service invocation and a service supporting local invocations is deployed locally, the In-Memory transport is always selected to optimize performance.

Web Services support

Today, when people talk about ESB they often emphasize Web Services processing. Technically, JBoss ESB has a number of Web Service-based components for exposing and invoking Web Service endpoints (i.e. SOAP onto and off of the bus):

SOAPProcessor action supports invocation of a JBossWS hosted Web Service endpoint through any JBossESB-hosted listener. This means the ESB can be used to expose Web service endpoints for any ESB services. This implementation is based on a thin Service Wrapper Web service (e.g. a JSR 181⁴ implementation) that wraps calls to the target Service. This also means that these Services are invocable over any transport channel supported by the ESB (http, ftp, jms etc.). SOAP processor is configured to point to the Web service endpoint (java class annotated with JAXWS annotations) that does the actual processing of the SOAP request.

SOAPClient action uses the WISE⁵ Client Service to generate a JAXWS client class and call the target service. It is using the URL of the Web Service WSDL to dynamically generate a JAXWS client to invoke a web service.

In our implementations, we found this support rather cumbersome to use, and as a result employed the following strategy:

• For consuming web services we generated web services clients from WSDL available to us (Axis 1, Axis 2 and/or JBoss WS can be used for this operation)
• For exposing ESB services we used JBoss WS (for SOAP) and RestEasy (for REST) to expose org.jboss.soa.esb.client.ServiceInvoker and invoking the required service. Such an approach is equivalent (architecturally) to the usage of the SOAPProcessor action, but a little bit more generic, allowing the service functionality to be exposed as SOAP, REST or any other desired interface.

JBoss ESB tooling

Configuration of the service listeners and service execution pipelines is done using XML. This XML configuration is contained in 3 files, which are required for an ESB service⁶:

• File jbm-queue-service.xml contains definitions of MBeans associated with each of the transports (typically queues) used by services⁷
• File deployment.xml contains the list of all transports that service relies on. The content of this file is used by a service (ESB) deployer to validate that all dependencies are satisfied.
• File jboss-esb.xml contains definitions of all the listeners and service pipelines.

While files jbm-queue-service.xml and deployment.xml have to be created straight in XML, JBoss provides tooling to simplify the creation, deployment and maintenance of ESB projects⁸. Figure 3 presents a snapshot of jboss-esb.xml editor.

Figure 3 JBoss ESB tooling

The tooling provides graphical support for adding and configuring Listeners, Services and individual actions (including action parameters).

Data transformation with Smooks

One of the most important service actions is data transformation. JBoss ESB leverages Smooks⁹ for implementation of this functionality. The basic principal of Smooks is to take a data source of some kind and generate an Event Stream from it. It then applies the Visitor pattern logic to this stream in order to produce a different kind of result.¹⁰ Many different data Source and Result types are supported, meaning many different transformation types are supported, including (but not limited to):

• XML to XML
• XML to Java
• Java to XML
• Java to Java
• EDI to XML
• EDI to Java
• Java to EDI
• CSV to XML
• CSV to ...

In order to simplify the use of Smooks, JBoss ESB provides a specialized action<¹¹ that can be configured to invoke Smooks– based transformation as part of the service pipeline.

Smooks Tooling

JBoss ESB tooling provides a graphical editor allowing visual definition of Smooks Transformations.

Figure 4 Smooks Graphical Editor

Once the transformation is created, Smooks’ Execution Report (Figure 5) allows a visual review of the sequence of steps that have been executed during transformation. This visual review can significantly simplify the debugging of Smooks transformations.

Figure 5 Smooks Execution report

<a>
  <b>

	<c>12345</c>
  </b>
  <d>

	<c>12345</c>
  </d>
</a>

JBPM

JBoss jBPM is a flexible, extensible framework for process languages¹³ . jPDL is one process language that is built on top of that common framework. It is a process language that expresses business processes graphically in terms of tasks, wait states for asynchronous communication, timers, automated actions and other components.

jPDL has minimal dependencies and can be used as easy as using a Java library. In addition, it can also be used in environments where extreme throughput is crucial by deploying it on a J2EE clustered application server. It can be configured with any database and can be deployed on any application server.

Unlike BPEL, which is tightly coupled to Web Services and Web services invocation (in BPEL, every activity has to be implemented as a Web service¹⁴ ), JBPM is more of a component framework¹⁵ , allowing direct invocations of the Java handlers (similar to an ESB service execution pipeline)

The main constructs defined by JPDL includes nodes and execution context. JPDL defines the following types of nodes:

• Start Node – Start of the process
• Task – Human Activity
• State - Wait States
• Node – Execution of Custom Code
• Decision – Decision based on process variables
• Fork – Splits execution along multiple paths
• Join – Combines multiple execution paths
• Transitions – Linkages between nodes
• SuperState – Aggregation of multiple nodes
• End Node – Completion of the process

Execution context is somewhat analogous to the HTTP session and contains a set of named objects – similar to the ESB message body, defined previously. Any node can access content of the execution context and read/write variables contained in the context. The execution state is persisted in the database - when the JPDL process is in the wait state, its context (context variables) is stored in the database; when the process is resumed, context variables are hydrated from the database and context is recreated.

A workhorse of JPDL process (in the case of service orchestration) is Node. A typical Node executes an action handler – a class implementing the org.jbpm.graph.def.ActionHandler interface. Similar to service action handlers, node action handlers are configurable, although they use different configuration approaches. Any public/private variable defined in an action handler implementation class can be configured through process definition.

A Decision node is another type of node that supports custom implementations. Functionality of this node is implemented by a class that implements the org.jbpm.graph.node.DecisionHandler interface. A Decision Handler can be configured the same way as the Action handler when implementing Node functionality.

A parallel execution in JPDL is implemented using Fork/Join nodes. All the paths connecting Fork and Join nodes will be implemented in parallel.

One of the things widely used in services orchestration and is missing in JPDL is support for loops. A sequential loop can be easily implemented using Decision node, evaluating loop variable. A more complex (but common in the service orchestration) scenario is parallel loop – Fork/Join with the run time calculation of the amount of transition paths. There is no specialized support for this pattern in JPDL, but it was easy enough to create two custom handlers, implementing this functionality (Listing 1, Listing 2 in the Appendix).

Implementation of these handlers is based on JPDL Fork/Join implementation. Start handler creates several (controlled by the variable with the name LoopCount) tokens and starts all of them. End handler waits until all child tokens are completed and then transitions to the next node.

Exception handling

Process execution can encounter exceptions. JPDL allows association of an exception handler with every process node. An exception handler is a kind of action handler, which is invoked when exception occurs during the node execution. An example of a simple exception handler is presented in in the Appendix. This exception handler prints exception information and then transitions to the node specified in the destination variable.

Process Deployment

JPDL process definitions are deployed not as JBoss applications, but rather in the process database.

JBPM/ESB integration

Both JBoss ESB and JBPM are very powerful software platforms, but their power significantly increases through ESB/JBPM integration¹⁶ .

There are two types of ESB/JBPM integrations:

• Exposing business processes as services
• Invoking services from business processes.

JBoss/JBPM integration provides a specialized action handler BpmProcessor action, which uses the jBPM command API to make calls into jBPM. This action takes a process definition name and creates and starts a new instance of a process. A process itself in this case is executed asynchronously on a separate thread, which means that service returns the reply to the invoker while process is running.

The integration also implements two jBPM action handler classes - EsbActionHandler and EsbNotifier. The EsbActionHandler is a request-reply type action, which drops a message on a Service bus and then waits for a response. The architecture for this integration is presented at Figure 6. EsbActionHandler sends a request message to the user service and puts a process token of execution in the wait state. When an execution of a user service completes, it invokes a special JBPM service, which signals waiting process token to continue.

Figure 6 JBPM/ESB integration architecture

As a result, such integration provides asynchronous execution support for a process with multiple tokens (e.g. in the case of fork/join or parallel loop execution see above).

An EsbNotifier, on the other hand, only drops a message on a Service and continues its processing. The interaction with JBossESB is asynchronous in nature and does not block the process instance while the Service executes.

Synchronous Invocation of the process

As we have mentioned above, the BpmProcessor invokes business processes asynchronously, which is not always desirable scenario in the service orchestration; the result of the process execution might be used as a service reply. Additional support is provided by ESB/JBPM integration to implement this scenario¹⁷.

A BpmProcessor action supports an additional configuration parameter – reply-to-originator. If this parameter is set to true, BPMProcessor stores ReplyTo information of the service invoker in JBPM execution context of the newly created process instance. This ReplyTo information can then be used by an EsbNotifier to deliver the process execution result by specifying reply-to-originator as the notifier’s destination. In order for this mechanism to work, the service’s invoking process has to be defined with MEP Oneway.

Exception handling

When a service in invoked from a business process, an exception can occur not only in the node execution (service invocation), but also in the execution of the invoked service. This situation requires a special exception handler. This handler can be defined as part of a BpmProcessor action and allows to control transition from service invocation – different transition in the case of success or error.

JBPM Tooling

JPDL is an XML-based language and consequently is expressed in XML. In order to simplify creation and analysis of business processes, JBoss provides an Eclipse plugin for visual creation and manipulation of JPDL processes (Figure 7).

This editor allows for both Graphical and XML-based view of business processes and also supports deployment from Eclipse project directly to the JBoss server.

Figure 7 JPDL editor

Overall Assessment

The combination of JBoss ESB/jBPM provides a very powerful, extensible and flexible platform for the creation of service-oriented solutions based on the existing enterprise assets. It provides all the major components for implementing such solutions, including:

• Flexible ESB platform, allowing easy combination of existing functionality with additional business and infrastructure processing required for service implementation (service pipelines).
• Powerful transformation engine (Smooks), simplifying implementation of transformation from proprietary data models (provided by existing functionality) to semantic data models used by the solution. Tooling that supports visual definition of the transformation further simplifies such implementations.
• Lightweight extensible jBPM implementation facilitates building composite services that are based on existing NAVTEQ functionality.

The JBoss middleware assessment, presented in this article, is based on a several VMS solutions prototypes built within NAVTEQ. The prototype was based on the domain-specific model for management users, locations and routes and included several ESB services, implemented as both wrappers of existing Web services, and services implemented directly inside the SOA platform itself. It also included several composite (and hierarchically composite¹⁸) services implemented using jBPM.

We are also currently in the process of evaluating performance and scalability of JBoss middleware – based solutions and usage of JBoss Operations Network¹⁹ (JON) for SOA solutions monitoring and management.

Acknowledgements

I am thankful to my NAVTEQ colleagues, especially Robert Camp, Ian Mondragon and Jeffrey Herr and JBoss solution architects, especially Ray Ploski and Aaron Pestel for implementing prototypes and describing their results.

Appendix. Code

package com.navteq.jbpm.parallel;


import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.jbpm.graph.def.ActionHandler;
import org.jbpm.graph.def.Node;
import org.jbpm.graph.def.Transition;
import org.jbpm.graph.exe.ExecutionContext;
import org.jbpm.graph.exe.Token;

 /**
  * specifies configurable parallel execution behaviour.
  * 
  */

public class ParallelStart implements ActionHandler {

	private static final long serialVersionUID = 1L;

	// The name of the variable holding loop count
	private String loopCount;
	// The name of the variable current count (stored in the token variable scope)
	private String currentCount;
	// The list of the array variables. Each array has to be of the of the loop count 
 	// size. An appropriate element is stored in the token context scope
	private Map<String, String> variables


	public void execute(ExecutionContext context) {

		// Load loop counter
		int counter;
		try {
			counter = (Integer)context.getContextInstance().getVariable(loopCount);
		}
		catch(Exception e){
			System.out.println("Can't load loop counter ");
			e.printStackTrace();
			return;
		}

		// Build splitter map
		Map<String, Object[]> splitMap = null;
		if((variables != null) && (variables.size() > 0)){
			splitMap = new HashMap<String, Object[]>();
			Set <String> keys = variables.keySet();
			for(String v : keys){
				try{
					Object[] values = (Object[])context.getContextInstance().getVariable(v);
					splitMap.put(v, values);
				}
				catch(Exception e){
					System.out.println("Can't load values for variable " + v);
					e.printStackTrace();
					continue;
				}
			}
		}

		// Get default transition
		Token token = context.getToken();
		Node currentNode = token.getNode();
		Transition tr = currentNode.getDefaultLeavingTransition();

		// execute default transition loop times
		for(int i = 0; i < counter; i++){
			Token childToken = new Token(token, tr.getName()+ i);
			ExecutionContext childExecutionContext = new ExecutionContext(childToken);
			childExecutionContext.getContextInstance().setVariable(FONT COLOR="#0000FF">currentCount, i, childToken);
			if(splitMap != null){

				// Populate variables
				Set<String> keys = splitMap.keySet();
				for(String v : keys){
					try{
						Object value = splitMap.get(v)[i];
						childExecutionContext.getContextInstance().setVariable(variables.get(v), value, childToken);
					}
					catch(Exception e){
						System.out.println("Can't load " + i + " element for variable " + v);
						e.printStackTrace();
						continue;
					}
				}
			}

			// Start token execution
			currentNode.leave(childExecutionContext);
		}
	}
}

Listing 1 Start Parallel execution handler

package com.navteq.jbpm.parallel;

import java.lang.reflect.Array;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import org.hibernate.LockMode;
import org.hibernate.Session;
import org.jbpm.JbpmContext;
import org.jbpm.graph.def.ActionHandler;
import org.jbpm.graph.exe.ExecutionContext;
import org.jbpm.graph.exe.Token;

public class ParallelEnd implements ActionHandler {

	private static final long serialVersionUID = 1L;

	/** specifies wether what type of hibernate lock should be acquired.  null value defaults to LockMode.force */
	String parentLockMode;
	// The name of the variable holding loop count
	private String loopCount;
	// The name of the variable current count (stored in the token variable scope)
	private String currentCount;
	// The list of the array variables. Each array has to be of the of the loop count
	// size. An appropriate element is stored in the token context scope
	private Map<String, String> variables;

	public void execute(ExecutionContext context) {

		// Get current token
		Token token = context.getToken();
		boolean isAbleToReactivateParent = token.isAbleToReactivateParent();

		if (!token.hasEnded()) {
			token.end(false);
		}

		// if this token is not able to reactivate the parent, 
 		// we don't need to check anything
		if ( isAbleToReactivateParent ) {

			// the token arrived in the join and can only reactivate 
 			// the parent once
			token.setAbleToReactivateParent(false);
			Token parentToken = token.getParent();
			// Build replies array
			if((variables != null) && (variables.size() > 0)){
				// current counter
				int current = (integer)context.getContextInstance().getVariable(currentCount,token);
				// for every variable
				Set<String> keys = variables.keySet();
				for (String v : keys) {
					Object value = context.getContextInstance().getVariable(v, token);
					String res = variables.get(v);
					Object[] values = (Object[]) context.getContextInstance().getVariable(res);
					if (values == null) {
						// Get array size
						int loopSize = (integer) context.getContextInstance().getVariable(loopCount);
						// Get variable type
						Class t = value.getClass();
						// Create an array of a variable type
						values = (Object[])Array.newInstance(t, loopSize);
					}
					values[current] = value;
					context.getContextInstance().setVariable(res, values);
				}
			}

			if ( parentToken != null ) {

				// Get context and lock session
				JbpmContext jbpmContext = context.getJbpmContext();
				Session session = (jbpmContext != null ? jbpmContext.getSession() : null);
				if (session != null) {
					LockMode lockMode = LockMode.forCE;
					if (parentLockMode!= null) {
						lockMode = LockMode.parse(parentLockMode);
					}
					session.lock(parentToken, lockMode);
				}

				// Check if we can reactivate parent
				boolean reactivateParent = mustParentBeReactivated(parentToken, parentToken.getChildren().keySet().iterator() );

				// if the parent token needs to be reactivated from this join node
				if (reactivateParent) {

					// write to all child tokens that the parent is already reactivated
					Iterator iter = parentToken.getChildren().values().iterator();
					while ( iter.hasNext() ) {
						((Token)iter.next()).setAbleToReactivateParent( false );
					}

					// write to all child tokens that the parent is already reactivated
					ExecutionContext parentContext = new ExecutionContext(parentToken);
					token.getNode().leave(parentContext);
				}
			}
		}
	}

	public boolean mustParentBeReactivated(Token parentToken, Iterator childTokenNameIterator) {

		boolean reactivateParent = true;
		while ( (childTokenNameIterator.hasNext())
				&& (reactivateParent) ){
			String concurrentTokenName = (String) childTokenNameIterator.next();

			Token concurrentToken = parentToken.getChild( concurrentTokenName );

			if (concurrentToken.isAbleToReactivateParent()) {
				reactivateParent = false;
			}
		}
		return reactivateParent;
	}
}

Listing 2 End Parallel execution handler

package com.navteq.jbpm.actionHandlers;

import org.jbpm.graph.def.ActionHandler;
import org.jbpm.graph.def.Node;
import org.jbpm.graph.exe.ExecutionContext;
import org.jbpm.graph.exe.Token;

public class ExceptionActionHandler implements ActionHandler {

	private static final long serialVersionUID = 1L;

	private String destination;

	@Override
	public void execute(ExecutionContext context) throws Exception {

		Token token = context.getToken();
		Node sourceNode = token.getNode();
		Throwable throwable = context.getException();
		System.out.println("Caught Exception " + throwable.getMessage() + " in node " + sourceNode.getName());
		Node targetNode = context.getProcessDefinition().getNode(destination);
		token.setNode(targetNode);
		token.signal();
	}

Listing 3 Exception handler