This section presents some definitions that are useful to understand before discussing activities in more detail.

Activity inputs receive streams of blocks that are simply Java objects. Special list begin and list end markers are often inserted into the data streams to group several blocks together. The term logical value is used to refer to either:

An activity's inputs are said to be matched if the activity expects each input to receive the same number of logical values. An example of a matched activity is ObtainFromFTP. This activity has three inputs: filename, host and the optional passiveMode. The activity expects the same number of logical values to be passed to the filename and host inputs and also to the passiveMode input if it is specified. The activities released with OGSA-DAI have been carefully designed to ensure they are matched.

An activity is said to be iterative if the activity performs the same processing repeatedly over different sets of inputs. An example of an iterative activity is ObtainFromFTP. This activity repeatedly reads one value from the filename input and one value from the host inputs, obtains the file from the FTP server specified by the host input and writes the data to the activity's output. The activity will continue to repeat this operation until it receives no more input data. The activities released with OGSA-DAI have been carefully designed to ensure they are iterative.

An activity input is said to be a multiple input if the activity can have multiple occurrences of that input. For example, consider the ListConcatenate activity. This activity concatenates multiple lists into a single list. The activity has a multiple input called input and an output called output. Figure 24.1, “Activity with multiple inputs of the same name.” shows an instance of this activity with three occurrences of the input input. The activity reads data from each of these inputs much like any other input. The ListContatenate activity processes each input in turn reading a list from each and writing the values to the output. The use of a multiple input allows the activity to concatenate as many inputs as the client specifies in the workflow.

ListContantenate activity with multiple input inputs.

Figure 24.1. Activity with multiple inputs of the same name.

An activity output is said to be a multiple output if the activity can have multiple occurrences of that output. For example, consider the TupleSplit activity. This activity reads a tuple list and distributes the tuples in the list evenly between a number of outputs. The activity has an input called data and an multiple output called result. Figure 24.2, “Activity with multiple outputs of the same name.” shows an instance of this activity with multiple occurrences of the result output. The activity writes the tuples it receives as input to the outputs in a round robin manner. The use of a multiple output allows the activity to distribute the tuples over as much outputs as the client specifies in the workflow.

TupleSplit activity with multiple result outputs.

Figure 24.2. Activity with multiple outputs of the same name.

A server side activity implementation class must implement the uk.org.ogsadai.activity.Activity interface. Typically this is done by extending one of four abstract base classes shipped with OGSA-DAI. These classes are:

	Note
	The first three abstract classes are extended the ActivityBase class which is responsible for providing access to input and output pipes, activity contracts and simple validation functionality. They only aim at providing a more convenient way to fragment and manipulate the activity processing.

The MatchedIterativeActivity base class is the most commonly used. It should be used in most cases when the activity is iterative and has matched inputs. This base class handles much of the functionality general to these types of activities including the detection and error handling of unmatched input data.

If an activity is both iterative and has matched inputs but at least one of these inputs is a multiple input then the activity should extend the MatchedIterativeMultipleInputActivity base class. This class is very similar to MatchedIterativeActivity but includes extra functionality for multiple inputs.

If the activity is iterative but has only a single input and you do not wish the base class to perform any error detection or processing on that input then the activity should extend IterativeActivity. An example of this is the Tee activity which is most easily implemented by iterating and processing each input object in turn rather than dealing with logical values as it would be required to do it extended MatchedIterativeActivity.

If an activity is not iterative or it has no inputs then none of the standard patterns apply and hence the activity should extend the ActivityBase class. An example of this is the GetAvailableTables activity that has no inputs.

The MatchedIterativeActivity and MatchedIterativeMultipleInputActivity classes make use of the ActivityInput interface to specify the properties of an activity's inputs. There are a number of very useful classes that implement this interface. These are listed here:

All these classes are in the uk.org.ogsadai.activity.io namespace.

When processing activities can throw three types of exception. These are:

The BlockReader and BlockWriter interfaces can throw exceptions when an activity reads and writes data. These exceptions are:

The PipeClosedException exception is thrown by the write method of BlockReader if the consuming activity at the other end of the pipe has closed the pipe indicating the activity does not wish to receive any more data. In this situations most activities simply stop producing the data.

The PipeIOException exception is thrown if there is an internal problem during reading from, or writing to, a pipe. Typically this exception is simply mapped to an ActivityPipeProcessingException exception as shown in the example activities.

The PipeTerminatedException exception is thrown if the activity has been terminated. Activities are typically notified that the are to terminate by receiving this exception. The activity should stop processing as soon as possible and throw the ActivityTerminatedException exception.

If an activity to interact with an OGSA-DAI resource then it must implement the uk.org.ogsadai.activity.extension.ResourceActivity activity extension interface.

A client using the activity in a workflow will specify the ID of the resource with which the activity is to interact. On encountering any activity that implements ResourceActivity the activity framework will provide the activity with the resource via the ResourceActivity interface's setTargetResourceAccessor(ResourceAccessor resourceAccessor) method.

This provides the activity with an resource accessor corresponding to the resource at which the activity is targeted. A resource accessor is a resource configured with a security context related to the client who submitted the original request. This is one way in which OGSA-DAI ensures that clients can only interact with resources in the ways that OGSA-DAI administrators allow (see Section 30.6.3, “Creating resource accessors” for information on resource accessors from a data resource developers perspective).

Each OGSA-DAI data resource implementation class has an associated sub-class of resource accessor. For example:

Activities that implement ResourceActivity must also support a method getTargetResourceAccessorClass(). This allows them to inform the activity framework of the types of resource with which they may be used by providing the name of the resource accessor interface they expect. The activity framework can then validate that the resource specified by a client is compatible with the activity and so, for example, allow SQLQuery to be targeted at a relational resource but not at an XMLDB data resource, for example. Activities return the name of a sub-interface of ResourceAccessor corresponding to the resource accessor supported by the data resources with which they can interact.

As an example SQLQueryActivity implements the methods of ResourceActivity as follows:

public Class getTargetResourceAccessorClass()
{
    return JDBCConnectionProvider.class;
}

and

public void setTargetResourceAccessor(final ResourceAccessor targetResource)
{
    mResource = (JDBCConnectionProvider) targetResource;
}

If the activity needs to read/write objects from session resources then the activity must implement the uk.org.ogsadai.activity.extension.SessionActivity activity extension interface.

On encountering any activity that implements the SessionActivity interface the activity framework will provide the activity with a reference to the session specified along with that activity in the request.

If the specified session exists then the activity framework invokes the SessionActivity.setSession method that gives the activity the specified session resource. This method will be called before any of the processing methods are called. Typically the method will simply store the session resource in a local variable for use later. For example:

public void setSession(SessionResource session)
{
    mSession = session;
}

See uk.org.ogsadai.activity.delivery.ObtainFromSession.

If the activity needs to access the security context then the activity must implement the uk.org.ogsadai.activity.extension.SecureActivity activity extension interface. This interface contains the setSecurityContext method that gives the activity the security context. This method will be called before any of the processing methods are called. Typically the method will simply store the security context in a local variable for use later. For example:

public void setSecurityContext(SecurityContext context)
{
    mSecurityContext = context;           
}

See uk.org.ogsadai.activity.delivery.DeliverToGFTP.

When the activity is to execute callable tasks - e.g. sub-workflows the uk.org.ogsadai.activity.extension.TaskProcessingActivity interface needs to be implemented. In this case, the Task Processing Extension will have to be set by setting the task factory and task processing service, e.g.

public void setTaskProcessingExtension(
    final AuthorizingTaskFactory taskFactory, 
    final TaskProcessingService processingService)
{
	mTaskFactory = taskFactory;
    mTaskProcessingService = processingService;
}

See uk.org.ogsadai.activity.sql.SQLBag.

Depending upon an application-specific requirement there might be a need or a desire to write an activity that can create application-specific data resources. This then allows clients to request the creation of these resources in their workflows.

OGSA-DAI ships with example resource creation activities including those to create data sources and data sinks (CreateDataSource and CreateDataSink) and, of more direct relevance to this section, CreateResourceGroup which creates OGSA-DAI resource group data resources.

When the activity needs to create new data resources the following steps need to be done.

src/core/resource/uk/org/ogsadai/resource/dataresource/group/

org.some.com.MY_RESOURCE_TEMPLATE

deploy/server-config/resourceTemplates/uk.org.ogsadai.RESOURCE_GROUP_TEMPLATE 

uk.org.ogsadai.activity.extension.ResourceFactoryActivity
uk.org.ogsadai.activity.extension.ResourceManagerActivity

public void setResourceManager(ResourceManager resourceManager)
{
    mResourceManager = resourceManager;        
}

public void setResourceFactory(ResourceFactory resourceFactory)
{
    mResourceFactory = resourceFactory;        
}

ResourceGroupDataResource resourceGroup = 
  (ResourceGroupDataResource)mResourceFactory.createDataResource(OGSADAIConstants.RESOURCE_GROUP_TEMPLATE); 

MyDataResource myResource = 
  (MyDataResource)mResourceFactory.createDataResource(new 
      uk.org.ogsadai.resource.ResourceID("org.some.com.MY_RESOURCE_TEMPLATE"); 

// Get the resource state.
ResourceGroupDataResourceState state = 
   resourceGroup.getResourceGroupDataResourceState(); 

// Set the ID of the new ResourceGroup...

state.getDataResourceState().setResourceID(groupResourceID); 

// ... and adds the member resources... 

String resourceID = null; 
while ((resourceID = (String)rIDs.nextValue()) != null) 
{ 
     state.addResourceID(new ResourceID(resourceID)); 
} 

mResourceFactory.addResource(groupResourceID, resourceGroup); 

// Activity input - ID for the new resource.
public static final String INPUT_RESOURCE_ID = "resourceID";
// Activity input - ID for some input from the client used to
// configure the resource. 
public static final String INPUT_CONFIG_ID = "someConfig";
// Activity output - ID for the new resource.
public static final String OUTPUT = "output";
// Resource template ID.
public static final ResourceID TEMPLATE_ID =
    new ResourceID("org.some.com.MY_RESOURCE_TEMPLATE"); 

protected ActivityInput[] getIterationInputs()
{
    // Declare our two inputs - the resource ID and configuration
    // parameter.
    return new ActivityInput[] 
    {
        // Input 0
        new TypedOptionalActivityInput(INPUT_RESOURCE_ID, String.class),
        // Input 1
        new TypedActivityInput(INPUT_CONFIG_ID, String.class)
    };
}

protected void preprocess() 
    throws ActivityUserException, ActivityProcessingException, ActivityTerminatedException
{
    validateOutput(OUTPUT);
    mOutput = getOutput();
}

protected void processIteration(Object[] iterationData)
    throws ActivityUserException, ActivityProcessingException, ActivityTerminatedException
{
    // Get the configuration value.
    String someConfig = (String)iterationData[1];
    ResourceID resourceID;
    if (iterationData[0] == null)
    {
        // No value was given for the optional resource ID input
        // so ask the resurce manager to create a unique resource ID.
        resourceID = mResourceManager.createUniqueID();
    }
    else
    {
        // Use the resource ID given by the client.
        resourceID =  new ResourceID((String)iterationData[0]);
    }
    try
    {  
        // Create a new instance of MyDataResource using a 
        // resource template.
        MyDataResource myDataResource
            (MyDataResource)mResourceFactory.createDataResource(TEMPLATE_ID);

        // Get its state and set its ID,
        MyDataResourceState state = myDataResource.getMyDataResourceState();
        state.getDataResourceState().setResourceID(resourceID);
        // Set the configuration.
        state.setSomeConfig(someConfig);

        // Add the resource to the resource manager using the
        // resource factory.
        mResourceFactory.addResource(resourceID, myDataResource);
        // Output the new resource ID.
        mOutput.write(resourceID.toString());
    }
    catch (PipeClosedException e)
    {
        // No more input wanted
        iterativeStageComplete();
    }
    catch (PipeTerminatedException e)
    {
        throw new ActivityTerminatedException();
    }
    catch (PipeIOException e)
    {
        throw new ActivityPipeProcessingException(e);
    }
    catch (ResourceIDAlreadyAssignedException e)
    {
        // Wrap the resource ID already assigned error as a user exception
        throw new ActivityUserException(e);
    }
    catch (Exception e)
    {
        throw new ActivityProcessingException(e);
    }
}

See uk.org.ogsadai.activity.management.CreateResourceGroupActivity.

When the activity needs to interact with the resource manager then the uk.org.ogsadai.activity.extension.ResourceManagerActivity can be implemented. The resource manager needs to be set:

public void setResourceManager(ResourceManager resourceManager)
{
	mResourceManager = resourceManager;        
}

Activities that might want to interact with a resource manager include those that create new data resources (e.g. CreateResourceGroup), destroy existing resources or query the state of the OGSA-DAI server.

See uk.org.ogsadai.activity.management.CreateResourceGroupActivity.

If an activity has server-side configuration properties then the uk.org.ogsadai.activity.extension.ConfigurableActivity interface has to be implemented. This allows the configuration from the OGSA-DAI persistence and configuration components to be given to the activity via a method:

public void configureActivity(KeyValueProperties properties);

An example of this interface is uk.org.ogsadai.activity.deliver.DeliverToSMTP activity. In this activity, the SMTP server that the activity will be using should be set by the OGSA-DAI administrator and therefore it will be a configuration parameter of the activity. More details about how an activity is configured can be found in Section 17.2.2, “Activity specification file”.

So if the configuration parameter is dai.smtp.server then the activity can define a key for this property.

private static final Key SMTP_SERVER = new Key("dai.smtp.server"); 

and the method configureActivity(KeyValueProperties properties) will be implemented like:

public void configureActivity(KeyValueProperties properties)
{
	mServerName = (String)properties.get(SMTP_SERVER);        
}

where server name is the actual value of key SMTP_SERVER.

Other possible interfaces can be found under uk.org.ogsadai.activity.extension.

There are two types of activities in OGSA-DAI, those that target a resource and those that do not. If the activity targets a resource then the client-side activity class must extend BaseResourceActivity. If the activity does not target a resource then the client-side activity class must extend BaseActivity. Both of these classes are in namespace uk.org.ogsadai.client.toolkit.activity.

For each non-multiple input that an activity has the client-side activity class should provide a connectXInput method and, depending on the type of objects the input receives, zero or more addX methods, where X is the name of the input.

public void connectXInput(ActivityOutput output);

public void connectXInput(ActivityOutput output)
{
    mXInput.connect(output);
}

private ActivityInput mXInput;

mXInput = new SimpleActivityInput("X");

mXInput = new SimpleActivityInput("X", SimpleActivityOutput.OPTIONAL);

If an activity supports multiple inputs with the same name (activity input with multiple occurences) then we consider that every activity input is a group of SingleActivityInput objects. By default an activity input will consist of one SingleActivityInput. Therefore, in the case of multiple inputs we need to specify the number of SingleActivityInput objects. To faciliatate this we need:

For each output an activity has the client-side activity implementation should provide methods to access the output values if these output values are of a type that can be written to the request status ( String, char[], byte[], Boolean, Date, number objects as well as lists of any of these).

Typically the activity can output a stream of values not just one. Thus for each such output named X we provide two methods:

boolean hasNextX();

and

Y nextXXX();

where Y is the type of the objects corresponding to the output.

The following classification applies to the nextX method:

Of course and apart from the above rules, when giving the user access to the outputp, one has to have in mind that we need to provide the data to the user in a convinient way based on the type of the output data. On good example of this is the TupleToWebRowSetCharArrays activity that has an output called result that outputs list of char[]. This character data contains XML respresentations of tuple sets so the client-side activity provides the additional convenience method

ResultSet getResultAsResultSet()

that allows users to obtain the data in a convenient object.

If an activity supports multiple outputs with the same name (output with multiple occurences) then we consider that every activity output is a group of uk.org.ogsadai.client.toolkit.SingleActivityOutput objects. By default an activity output will consist of one SingleActivityOutput. Therefore in the case of multiple outputs we need to specify the number of SingleActivityOutput objects. To faciliatate this we need, for output X:

If an activity's output is connected to a deliverToRequestStatus activity so that the result comes back to the user then the data values written to that output can be obtained from the ActivityOutput object using the getDataValueIterator method. This is how we get hold of the data stream to implement the hasNextX and nextX methods described above.

If the activity (activity A) is not connected directly to the deliverToRequestStatus activity but there is another activity (activity B) in between that does not alter the data in a way that prevents activity A handling the output then activity A can still obtain the data that was written to he deliverToRequestStatus. To do this the application developer must tell activity A about the activity used to deliver the data to the request status. For example:

ActivityA activityA = new ActivityA();
ActivityB activityB = new ActivityB();
DeliverToRequestStatus deliverToRequestStatus = 
    new DeliverToRequestStatus();
activityB.connectInput(activityA.getOutput());
deliverToRequestStatus.connectInput(activityB.getOutput);
// Tell activity A's output how to get the data from the request status
activityA.getOutput().setDeliverToRequestStatusActivity(
    deliverToRequestStatus);

The base activity class will do most of the activity input and output validation. If however you need to do validation that requires looking at several inputs or outputs at once (for example, if an activity must have one of inputs A or B but never both) then this logic can be coded in the validateIOState method.

Very few activities need this type of validation so typically the method is empty:

protected void validateIOState() throws ActivityIOIllegalStateException
{
}