The BPM (Business Process Management) specific activity events are used to record the lifecycle and state transitions that occur when a business process (associated with a description language such as BPMN2 or WS-BPEL) is executed within a runtime engine, in support of a business transaction.

These business processes tend to be "long running", in that they handle multiple requests and responses over a period of time, all being correlated to the same process instance. This means that activities generated as a result of this execution must also be correlated to \(i) the specific XA transaction in which they are performed, (ii) the process instance that holds their state information in the BPM engine, and (iii) the conversation associated with the particular business transaction.

This does not mean that all Activity Units the contain activity information from the BPM engine need to have all three types of correlation information. For example, the initial Activity Unit for a business process instance may identify (i) and (ii), which will establish a unique process instance id. A subsequent Activity Unit may then define the same process id for (ii), as well as a conversation id (iii) that can then be used to tie any Activity Unit relates with the process instance id to that conversation - i.e. all Activity Units with the same process instance id become directly or indirectly correlated to the conversation id that may only be declared in some of the Activity Units.

In a JEE environment, the Activity Collector is obtained from JNDI using the following code:

import org.overlord.rtgov.activity.collector.ActivityCollector;

....

javax.naming.InitialContext ctx=new javax.naming.InitialContext();

ActivityCollector activityCollector = (ActivityCollector)ctx.lookup("java:global/overlord-rtgov/ActivityCollector");

To start the scope, simply invoke the startScope method on the Activity Collector:

activityCollector.startScope();

If the application does not know whether a scope has already been started, and only wishes to start a single scope (i.e. as nested scopes are not supported), then the following guard can be used:

boolean started=false;
if (!activityCollector.isScopeActive()) {
    activityCollector.startScope();
    started = true;
}

The isScopeActive method returns a boolean value to indicate whether the scope was previously started. If true is returned, then this component is also responsible for stopping the scope. If false is returned, then it means the scope has already been started, and therefore the component should NOT invoke the endScope method.

To stop the scope, simply invoke the endScope method on the Activity Collector:

if (started) {
    activityCollector.endScope();
}

The Batched Activity Unit Logger is an abstract base class implementing the Activity Unit Logger interface. It provides the functionality to batch Activity Unit instances, and then forwarding them based on two properties:

This implementation can be explicitly initialized when used in an embedded environment. If used within a JEE environment, then the PostConstruct and PreDestroy annotations enable it to be implicit initialized and tidied up when the concrete component’s lifecycle is managed.

This implementation of the Activity Unit Logger interface is derived from the Batched Activity Unit Logger, and therefore will send activity information in a batch periodically based on the configured properties. When the batch of Activity Units are sent, this implementation forwards them to an implementation of the Activity Server interface, injected explicitly or implicitly into the logger.

The Activity Server will be discussed in a subsequent section of this document. However, this can be used to either send the events directly to the Activity Server component, if co-located within the same server, or via a remote binding. For example,

import org.overlord.rtgov.activity.collector.ActivityCollector;
import org.overlord.rtgov.activity.collector.activity.server.ActivityServerLogger;
import org.overlord.rtgov.activity.server.rest.client.RESTActivityServer;

.....

        RESTActivityServer restc=new RESTActivityServer();
        restc.setServerURL(_activityServerURL);

        ActivityServerLogger activityUnitLogger=new ActivityServerLogger();
        activityUnitLogger.setActivityServer(restc);

        activityUnitLogger.init();

        _collector.setActivityUnitLogger(activityUnitLogger);

This shows a situation where an embedded Activity Collector is being initialized with an Activity Server Logger, which uses the REST Activity Server client implementation.

The simpliest approach is to leverage CDI, for example within a JEE container, to directly inject the Activity Server implementation.

import org.overlord.rtgov.activity.server.ActivityServer;

....

@Inject
private ActivityServer _activityServer=null;

Once the reference to the Activity Server has been obtained, then call the store method to record a list of Activity Units.

import org.overlord.rtgov.activity.model.soa.RequestSent;
import org.overlord.rtgov.activity.model.ActivityUnit;

....

java.util.List<ActivityUnit> list=new .....;

RequestSent act=new RequestSent();
act.setServiceType(...);
...

list.add(act);

_activityServer.store(list);

The Activity Server can be accessed as RESTful service, e.g.

import org.codehaus.jackson.map.ObjectMapper;
import org.overlord.rtgov.activity.model.ActivityUnit;


.....

        java.util.List<ActivityUnit> activities=........
        java.net.URL storeUrl = new java.net.URL(....);   // <host>/overlord-rtgov/activity/store

        java.net.HttpURLConnection connection = (java.net.HttpURLConnection) storeUrl.openConnection();
        connection.setRequestMethod("POST");
        connection.setDoOutput(true);
        connection.setDoInput(true);
        connection.setUseCaches(false);
        connection.setAllowUserInteraction(false);
        connection.setRequestProperty("Content-Type", "application/json");

        java.io.OutputStream os=connection.getOutputStream();

        ObjectMapper mapper=new ObjectMapper();    // Use jackson to serialize the activity units
        mapper.writeValue(os, activities);

        os.flush();
        os.close();

        java.io.InputStream is=connection.getInputStream();

        byte[] result=new byte[is.available()];

        is.read(result);
        is.close();

The simpliest approach is to leverage CDI, as illustrated above, to obtain a reference to the Activity Server. Once the reference to the Activity Server has been obtained, then build the query specification based on the relevant criteria, and call the query method to retrieve the list of Activity Units.

import org.overlord.rtgov.activity.model.ActivityUnit;
import org.overlord.rtgov.activity.model.Context;
import org.overlord.rtgov.activity.server.QuerySpec;

....

QuerySpec qs=new QuerySpec()
                .setId(...)
                .setFromTimestamp(...)
                .setToTimestamp(...)
                .addContext(new Context(Context.CONVERSATION_ID,"txnId","123"));


java.util.List<ActivityUnit> list=_activityServer.query(qs);

The Activity Server can be accessed as RESTful service, e.g.

import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;
import org.overlord.rtgov.activity.server.QuerySpec;
import org.overlord.rtgov.activity.model.ActivityUnit;


.....

        QuerySpec qs=........
        java.net.URL queryUrl = new java.net.URL(....);   // <host>/overlord-rtgov/activity/query

        java.net.HttpURLConnection connection = (java.net.HttpURLConnection) queryUrl.openConnection();
        connection.setRequestMethod("POST");
        connection.setDoOutput(true);
        connection.setDoInput(true);
        connection.setUseCaches(false);
        connection.setAllowUserInteraction(false);
        connection.setRequestProperty("Content-Type", "application/json");

        java.io.OutputStream os=connection.getOutputStream();

        ObjectMapper mapper=new ObjectMapper();    // Use jackson to serialize the query spec
        mapper.writeValue(os, qs);

        os.flush();
        os.close();

        java.io.InputStream is=connection.getInputStream();

        java.util.List<ActivityUnit> activities = mapper.readValue(is, new TypeReference<java.util.List<ActivityUnit>>() {});

        is.close();