Receiving the status of tasks dispatched to or returned from the nodes

Note: as of JPPF 5.1 this extension supersedes "Receiving the status of tasks returning from the nodes", which is now deprecated.

Each time a set of tasks is dispatched to, or returns from a node, the driver emits a notification which encapsulates information about the set of tasks, alongs with detailed information on each of the tasks it contains. To receive these notifications, you must write a listener which implements the JobTasksListener interface, defined as:

public interface JobTasksListener extends EventListener {
  // Called when tasks from a job are dispatched to a node
  void tasksDispatched(JobTasksEvent event);
  // Called when tasks from a job return from the node
  void tasksReturned(JobTasksEvent event);
}

As we can see, the tasksReturned() method receives events of type JobTasksEvent, defined as follows:

public class JobTasksEvent extends EventObject {
  // Get the uuid of the job to which the tasks belong
  public String getJobUuid()
  // Get the name of the job to which the tasks belong
  public String getJobName()
  // Get the list of tasks that were dispatched or returned
  public List<ServerTaskInformation> getTasks()
  // Get the reason why the set of tasks was returned by a node
  public JobReturnReason getReturnReason()
  // Get the information on the node where the tasks were dispatched or returned
  public JPPFManagementInfo getNodeInfo()
}

The method getTasks() returns a list of ServerTaskInformation objects, providing details on individual tasks:

public class ServerTaskInformation implements Serializable {
  // Get the position of this task within the job submitted by the client
  public int getJobPosition()
  // Get the throwable raised during the processing of the task
  public Throwable getThrowable()
  // Get the number of times a dispatch of the task has expired
  public int getExpirationCount()
  // Get the maximum number of times the task can be resubmitted
  public int getMaxResubmits()
  // Get the number of times the task was resubmitted
  public int getResubmitCount()
}

The method getReturnReason() provides a high-level indication of why the tasks were returned, among the possible reasons defined in the JobReturnReason enum:

public enum JobReturnReason {
  // The tasks were normally processed by the node
  RESULTS_RECEIVED,
  // The processing of the tasks took longer than the specified dispatch timeout
  DISPATCH_TIMEOUT,
  // An error occurred in the node which prevented the normal execution of the tasks
  NODE_PROCESSING_ERROR,
  // An error occurred in the driver while processing the results returned by the node
  DRIVER_PROCESSING_ERROR,
  // The connection between node and server was severed before results could be returned
  NODE_CHANNEL_ERROR
}

The job return reason is only available for JobTasksListener.tasksReturned() notifications. Therefore, for a JobTasksListener.tasksDispatched() notification, getReturnReason() always returns null. The following code is an example listener which prints out its events to the console:

public class MyJobTasksListener implements JobTasksListener {
  public MyJobTasksListener() {
    System.out.println("in MyJobTasksListener()"); // displayed at driver startup time
  }

  @Override public void tasksDispatched(TaskReturnEvent event) {
    printEvent(event);
  }

  @Override public void tasksReturned(TaskReturnEvent event) {
    printEvent(event);
  }

  private void printEvent(JobTasksEvent event) {
    System.out.printf("tasksReturned() : name=%s, uuid=%s, reason=%s, node=%s,
      %d tasks=%s%n", event.getJobName(), event.getJobUuid(), event.getReturnReason(),
      event.getNodeInfo(), event.getTasks().size(), event.getTasks());
  }
}

and here is an example output:

tasksReturned() : name=my job, uuid=job_uuid, reason=RESULTS_RECEIVED,
node=JPPFManagementInfo[192.168.1.24:12001, type=node|MASTER, local=false, secure=false,
uuid=node_uuid], 1 tasks=[ServerTaskInformation[jobPosition=85, throwable=null,
expirationCount=0, maxResubmits=1, resubmitCount=0]]

Lastly, there are two ways to integrate a JobTasksListener with the JPPF driver:

1) Using the JPPFDriver API

This can be done by calling the method getJobTasksListenerManager() on the JPPFDriver singleton instance, from another driver plugin, add-on or extension. For instance, as in this driver startup class:

public class MyDriverStartup implements JPPFDriverStartupSPI {
  @Override
  public void run() {
    // get the object which manages task return listeners
    JobTasksListenerManager manager =
      JPPFDriver.getInstance().getJobTasksListenerManager();
    // register a new TaskReturnListener
    manager.addJobTasksListener(new JobTasksListener() {
      @Override
      public void tasksDispatched(JobTasksEvent event) { ... }
      @Override
      public void tasksReturned(JobTasksEvent event) { ... }
    });
  }
}

Note that you will need to have the jppf-server.jar file in your build path for this code to compile.

2) Using the Service Provider Interface (SPI)

To register the JobTasksListener as a standalone service, create a service definition file named "org.jppf.job.JobTasksListener" in the "META-INF/services" directory. In this file, add, for each of your listener implementations, a line containing the fully qualified name of the implementation class. For instance, if we defined two implementations MyJobTasksListener1 and MyJobTasksListener2 in the test1 and test2 packages, respectively, then the service definiton file should contain:

test1.MyJobTasksListener1
test2.MyJobTasksListener2

Note that, to work with the SPI, each implementation must have a no-args constructor, whether implicit or explicit.