How to use the connector API

1 Obtaining and closing a resource adapter connection

The J2EE connector is accessed via the JPPFConnection interface. This implies that any operation performed should follow these steps:

1. obtain a connection from the resource adapter connection pool
2. perform one or more JPPF-related operation(s)
3. close the connection

The following helper code illustrates how to obtain and release connections from the resource adapter:

import javax.naming.*;
import javax.resource.ResourceException;
import javax.resource.cci.ConnectionFactory;
  
public class JPPFHelper {
  // JNDI name of the JPPFConnectionFactory
  public static final String JNDI_NAME = "eis/JPPFConnectionFactory";

  // Obtain a JPPF connection from the resource adapter's connection pool
  public static JPPFConnection getConnection()
    throws NamingException, ResourceException {

    // Perform a JNDI lookup of the JPPF connection factory
    InitialContext ctx = new InitialContext();
    JPPFConnectionFactory factory;
    Object objref = ctx.lookup(JNDI_NAME);
    if (objref instanceof JPPFConnectionFactory) {
      factory = (JPPFConnectionFactory) objref;
    } else {
     factory = (JPPFConnectionFactory) javax.rmi.PortableRemoteObject.narrow(
        objref, ConnectionFactory.class);
    }
    // get a JPPFConnection from the connection factory
    return (JPPFConnection) factory.getConnection();
  }

  // Release a connection
  public static void closeConnection(JPPFConnection connection)
    throws ResourceException {
      connection.close();
  }
}

Please note that the actual JNDI name for the JPPF connection factory will vary depending on which application server is used:

• on Apache Geronimo: "jca:/JPPF/jca-client/JCAManagedConnectionFactory/eis/JPPFConnectionFactory"
• on JBoss 4-7: "java:eis/JPPFConnectionFactory"
• on Willdlfy 8: "java:/eis/JPPFConnectionFactory"
• on Websphere: "java:comp/env/eis/JPPFConnectionFactory"
• on all other supported servers: "eis/JPPFConnectionFactory"

2 Reset of the JPPF client

The method JPPFConnection.resetClient() will trigger a reset of the underlying JPPF client. This method enables reloading the JPPF configuration without having to restart the application server. Example use:

JPPFConnection connection = null;
try {
  connection = JPPFHelper.getConnection();
  connection.resetClient();
} finally {
  if (connection != null) JPPFHelper.closeConnection(connection);
}

As for JPPFClient.reset(), calling this method will not lose any already submitted jobs. Instead, the JCA connectior will resubmit them as soon as it is reset and server connections become available.

3 Submitting jobs

JPPFConnection provides two methods for submitting jobs:

public interface JPPFConnection extends Connection, JPPFAccessor {
  // Submit a job to the JPPF client
  // This method exits immediately after adding the job to the queue
  String submit(JPPFJob job) throws Exception;

  // Submit a job to the JPPF client and register the specified status listener
  // This method exits immediately after adding the job to the queue
  String submit(JPPFJob job, SubmissionStatusListener listener) throws Exception;
}

You will note that both methods actually perform an asynchronous job submission. They return a unique id for the the submission, which is in fact the job UUID. This id is then used to retrieve the job results and its status.

In the following example, a JPPF job is submitted asynchronously. The submission returns an ID that can be used later on to check on the job status and retrieve its results.

public String submitJob() throws Exception {
  JPPFConnection connection = null;
  try {
    // get a JPPF Connection
    connection = JPPFHelper.getConnection();
    // create a JPPF job
    JPPFJob job = new JPPFJob();
    job.add(new DemoTask());
    // Use the connection to submit the JPPF job and obtain a submission ID
    return connection.submit(job);
  } finally {
    // close the connection
    JPPFHelper.closeConnection(connection);
  }
}

4 Checking the status and getting the results of a job

Here, we check on the status of a job and process the execution results or the resulting error:

public void checkStatus(String submitId) throws Exception {
  JPPFConnection connection = null;
  try {
    connection = JPPFHelper.getConnection();
    // Use the connection to check the status from the submission ID
    SubmissionStatus status = connection.getSubmissionStatus(submitID);
    if (status.equals(SubmissionStatus.COMPLETE)) {
      // if successful process the results
      List<Task<?>> results = connection.getResults(submitID);
    } else if (status.equals(SubmissionStatus.FAILED)) {
      // if failed process the errors
    }
  } finally {
    JPPFHelper.closeConnection(connection);
  }
}

5 Cancelling a job

The J2EE allows cancelling a job by calling the method JPPFConnection.cancelJob(String submitId):

public void cancelJob(String submitId) throws Exception {
  JPPFConnection connection = null;
  try {
    connection = JPPFHelper.getConnection();
    // cancel the job
    connection.cancelJob(submitID);
  } finally {
    JPPFHelper.closeConnection(connection);
  }
}

6 Synchronous execution

It is also possible to execute a job synchronously, without having to code the job submission and status checking in two different methods. The JPPFConnection API provides the method waitForResults(String submitID), which waits until the job has completed and returns the execution results. Here is an example use:

// Submit a job and return the execution results
public List<Task<?>> submitBlockingJob() throws Exception {
  List<Task<?>> results = null;
  JPPFConnection connection = null;
  try {
    connection = JPPFHelper.getConnection();
    // create a new job
    JPPFJob job = new JPPFJob();
    job.setName("test job");
    // add the tasks to the job
    for (int i=0; i<5; i++) job.addTask(new MyTask(i));
    // submit the job and get the submission id
    String submitID = connection.submit(job);
    // wait until the job has completed
    results = connection.awaitResults(submitID);
  } finally {
    JPPFHelper.closeConnection(connection);
  }
  // now return the results
  return results;
}

Please note that, when using the synchronous submission mode from within a transaction, you must be careful as to how long the job will take to execute. If the job execution is too long, this may cause the transaction to time out and roll back, if the execution time is longer than the transaction timeout.

7 Using submission status events

With the J2EE connector, It is possible to subscribe to events occurring during the life cycle of a job. This can be done via the following two methods:

public interface JPPFConnection extends Connection, JPPFAccessor {
  // Add a status listener to the submission with the specified id
  void addSubmissionStatusListener(
    String submissionId, SubmissionStatusListener listener);

  // Submit a job to the JPPF client and register a status listener
  String submit(JPPFJob job, SubmissionStatusListener listener) throws Exception;
}

Note that submit(JPPFJob, SubmissionStatusListener) submits the job and registers the listener in a single atomic operation. As the job submssion is asynchronous, this ensures that no event is missed between the submission of the job and the registration of the listener.

The interface SubmissionStatusListener is defined as follows:

public interface SubmissionStatusListener extends EventListener {
  // Called when the status of a job has changed
  void submissionStatusChanged(SubmissionStatusEvent event);
}

Each listener receives events of type SubmissionStatusEvent, defined as follows:

public class SubmissionStatusEvent extends EventObject {
  // get the status of the job
  public SubmissionStatus getStatus()

  // get the id of the job
  public String getSubmissionId()
}

The possible statuses are defined in the enumerated type SubmissionStatus:

public enum SubmissionStatus {
  // the job was just submitted
  SUBMITTED,
  // the job is currently in the submission queue (on the client side)
  PENDING,
  // the job is being executed
  EXECUTING,
  // the job execution is complete
  COMPLETE,
  // the job execution has failed
  FAILED
}

Here is an exemple usage of the status listeners:

public void submitWithListener() throws Exception {
  JPPFConnection connection = null;
  try {
    connection = JPPFHelper.getConnection();
    JPPFJob job = new JPPFJob();
    job.addTask(new DemoTask(duration));

    // a status listener can be added at submission time
    String id = connection.submit(job, new SubmissionStatusListener() {
      public void submissionStatusChanged(SubmissionStatusEvent event) {
        String id = event.getSubmissionId();
        SubmissionStatus status = event.getStatus();
        System.out.println("submission [" + id + "] changed to '" + status + "'");
      }
    });

    // or after the job has been submitted
    connection.addSubmissionStatusListener(id, new SubmissionStatusListener() {
      public void submissionStatusChanged(SubmissionStatusEvent event) {
        String id = event.getSubmissionId();
        SubmissionStatus status = event.getStatus();
        switch(status) {
          case COMPLETE:// process successful completion
            break;
          case FAILED:// process failure
            break;
          default:
            System.out.println("job [" + id + "] changed to '" + status + "'");
            break;
        }
      }
    });
    List<Task<?>> results = connection.awaitResults(id);
    // ... process the results ...
  } finally {
    JPPFHelper.closeConnection(connection);
  }
}