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 Wildfly 8: "java:/eis/JPPFConnectionFactory"
• on Websphere: "java:comp/env/eis/JPPFConnectionFactory"
• on all other supported servers: "eis/JPPFConnectionFactory"

2 The JPPFConnectionFactory API

In addition to implementing the ConnectionFactory interface, JPPFConnectionFactory provides two methods:

public class JPPFConnectionFactory implements ConnectionFactory {
  // Determine whether there is a least one working connection to a remote JPPF driver
  public boolean isJPPFDriverAvailable()

  // Enable or disable local (in-JVM) execution of the jobs
  public void enableLocalExecution(final boolean enabled)
}

Basically, these methods offer the ability to know whether the J2EE connector is effectively connected to a JPPF driver, and to fallback to local execution when it is not the case.

An example usage of these APIs would be as follows:

JPPFConnection connection = null;
try {
  JPPFConnectionFactory factory = ...;
  boolean available = factory.isJPPFDriverAvailable();
  // enable local execution, based on the availability of a remote connection
  factory.enableLocalExecution(!available);
  connection = (JPPFConnection) factory.getConnection();
  // ... submit a job ...
} finally {
  if (connection != null) connection.close();
}

3 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.

4 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, JobStatusListener 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 the job UUID, which 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 return the job uuid
    return connection.submit(job);
  } finally {
    // close the connection
    JPPFHelper.closeConnection(connection);
  }
}

5 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 jobUuid) throws Exception {
  JPPFConnection connection = null;
  try {
    connection = JPPFHelper.getConnection();
    // Use the connection to check the status from the job uuid
    JobStatus status = connection.getJobStatus(jobUuid);
    if (status.equals(JobStatus.COMPLETE)) {
      // if successful process the results
      List<Task<?>> results = connection.getResults(jobUuid);
    } else if (status.equals(JobStatus.FAILED)) {
      // if failed process the errors
    }
  } finally {
    JPPFHelper.closeConnection(connection);
  }
}

6 Cancelling a job

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

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

7 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 awaitResults(String jobUuid), 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.add(new MyTask(i));
    // submit the job and get the submission id
    String jobUuid = connection.submit(job);
    // wait until the job has completed
    results = connection.awaitResults(jobUuid);
  } 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.

8 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 job with the specified uuid
  void addJobStatusListener(String jobUuid, JobStatusListener listener);

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

Note that submit(JPPFJob, JobStatusListener) 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 JobStatusListener is defined as follows:

public interface JobStatusListener extends EventListener {
  // Called when the status of a job has changed
  void JobStatusChanged(JobStatusEvent event);
}

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

public class JobStatusEvent extends EventObject {
  // get the status of the job
  public JobStatus getStatus()

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

The possible statuses are defined in the enumerated type JobStatus:

public enum JobStatus {
  // 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 JobStatusListener() {
      public void JobStatusChanged(JobStatusEvent event) {
        String uuid = event.getJobUuid();
        JobStatus status = event.getStatus();
        System.out.println("sjob [" + uuid + "] changed to '" + status + "'");
      }
    });

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