0.10/workflow/src/main/java/org/apache/oodt/cas/workflow/engine/TaskQuerier.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.oodt.cas.workflow.engine;

//JDK imports
import java.util.List;
import java.util.Vector;
import java.util.logging.Level;
import java.util.logging.Logger;

//OODT imports
import org.apache.oodt.cas.workflow.engine.processor.TaskProcessor;
import org.apache.oodt.cas.workflow.engine.processor.WorkflowProcessor;
import org.apache.oodt.cas.workflow.engine.processor.WorkflowProcessorHelper;
import org.apache.oodt.cas.workflow.engine.processor.WorkflowProcessorQueue;
import org.apache.oodt.cas.workflow.instrepo.WorkflowInstanceRepository;
import org.apache.oodt.cas.workflow.lifecycle.WorkflowLifecycle;
import org.apache.oodt.cas.workflow.lifecycle.WorkflowState;
import org.apache.oodt.cas.workflow.structs.PrioritySorter;
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.exceptions.InstanceRepositoryException;

/**
 * 
 * The purpose of this class is to constantly pop off tasks that are run to run
 * and made available by the {@link WorkflowProcessorQueue}, and then to set
 * their state to Executing (running Category), so they will be picked up on the
 * next WorkflowState change, and end up executing.
 * 
 * @author mattmann
 * @author bfoster
 * @version $Revision$
 * 
 */
public class TaskQuerier implements Runnable {

  private boolean running;

  private WorkflowProcessorQueue processorQueue;

  private List<WorkflowProcessor> runnableProcessors;

  private PrioritySorter prioritizer;

  private WorkflowInstanceRepository repo;
  
  private long waitSeconds;

  private static final Logger LOG = Logger.getLogger(TaskQuerier.class
      .getName());

  /**
   * Constructs a new TaskQuerier with the given {@link WorkflowProcessorQueue},
   * and with the associated {@link PrioritySorter} which acts as a sorter of
   * the runnable {@link WorkflowProcessor}s.
   * 
   * @param processorQueue
   *          The associated set of queued Workflow Tasks.
   * @param prioritizer
   *          The prioritizer to use to sort the ready-to-run Workflow Tasks.
   * 
   * @param repo
   *          The {@link WorkflowInstanceRepository} to save the state of
   *          WorkflowInstances.
   *          
   * @param waitSeconds The default amount of seconds to wait while dispositioning
   *        processors
   */
  public TaskQuerier(WorkflowProcessorQueue processorQueue,
      PrioritySorter prioritizer, WorkflowInstanceRepository repo, long waitSeconds) {
    this.running = true;
    this.processorQueue = processorQueue;
    this.runnableProcessors = new Vector<WorkflowProcessor>();
    this.prioritizer = prioritizer;
    this.repo = repo;
    this.waitSeconds = waitSeconds;
  }

  /**
   * Marches through the set of processors that are currently in the Processor
   * queue, checks to see if they are NOT in the done state, or if they are
   * currently in the holding state. If either of those are true, the processor
   * is popped off the queue, and then added to the runnableProcessors list (in
   * a synchronized fashion), and then their state is set to Executing
   * (category, running).
   * 
   * Finally the runnableProcessors list is sorted according to the given
   * {@link #prioritizer}.
   */
  public void run() {
    LOG.log(Level.FINE, "TaskQuerier configured with wait seconds: ["+this.waitSeconds+"]");
    while (running) {
      List<WorkflowProcessor> processors = processorQueue.getProcessors();
      List<WorkflowProcessor> processorsToRun = new Vector<WorkflowProcessor>();

      for (WorkflowProcessor processor : processors) {
        // OK now get its lifecycle
        WorkflowProcessorHelper helper = new WorkflowProcessorHelper(
            processor.getLifecycleManager());
        WorkflowLifecycle lifecycle = helper
            .getLifecycleForProcessor(processor);

        LOG.log(Level.FINE, "TaskQuerier: dispositioning processor with id: ["
            + processor.getWorkflowInstance().getId() + "]: state: "
            + processor.getWorkflowInstance().getState());

        if (!processor.isAnyCategory("done", "holding")
            && !processor.isAnyState("Executing")
            && processor.getRunnableWorkflowProcessors().size() > 0) {
          for (TaskProcessor tp : processor.getRunnableWorkflowProcessors()) {
            WorkflowState state = lifecycle.createState("WaitingOnResources",
                "waiting", "Added to Runnable queue");
            tp.getWorkflowInstance().setState(state);
            persist(tp.getWorkflowInstance());
            LOG.log(Level.INFO, "Added processor with priority: ["
                + tp.getWorkflowInstance().getPriority() + "]");
            processorsToRun.add(tp);
          }

          if (processorsToRun != null && processorsToRun.size() > 1) {
            prioritizer.sort(processorsToRun);
          }

          synchronized (runnableProcessors) {
            if (running)
              runnableProcessors = processorsToRun;
          }

        } else {
          // simply call nextState and persist it
          LOG.log(Level.FINE, "Processor for workflow instance: ["
              + processor.getWorkflowInstance().getId()
              + "] not ready to Execute or already Executing: "
              + "advancing it to next state.");
          processor.nextState();
          persist(processor.getWorkflowInstance());
        }
      }
      
      try{
        Thread.currentThread().sleep(waitSeconds*1000);
      }
      catch(InterruptedException ignore){}
      
    }
  }

  /**
   * @return the running
   */
  public boolean isRunning() {
    return running;
  }

  /**
   * @param running
   *          the running to set
   */
  public void setRunning(boolean running) {
    this.running = running;
  }

  /**
   * @return the runnableProcessors
   */
  public List<WorkflowProcessor> getRunnableProcessors() {
    return runnableProcessors;
  }

  /**
   * Gets the next available {@link TaskProcessor} from the {@link List} of
   * {@link #runnableProcessors}. Removes that {@link TaskProcessor} from the
   * actual {@link #runnableProcessors} {@link List}.
   * 
   * @return The next available {@link TaskProcessor} from the {@link List} of
   *         {@link #runnableProcessors}.
   */
  public TaskProcessor getNext() {
    if (getRunnableProcessors().size() == 0)
      return null;
    return (TaskProcessor) getRunnableProcessors().remove(0);
  }

  private synchronized void persist(WorkflowInstance instance) {
    if (this.repo != null) {
      try {
        if (instance.getId() == null
            || (instance.getId() != null && instance.getId().equals(""))) {
          // we have to persist it by adding it
          // rather than updating it
          repo.addWorkflowInstance(instance);
        } else {
          // persist by update
          repo.updateWorkflowInstance(instance);
        }
      } catch (InstanceRepositoryException e) {
        e.printStackTrace();
        LOG.log(Level.WARNING, "Unable to update workflow instance: ["
            + instance.getId() + "] status to [" + instance.getState().getName()
            + "]. Message: " + e.getMessage());
      }
    }
  }

}