/*
    Copyright (C) 2006 Grid Systems, S.A.
   
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.
   
    This library is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   Lesser General Public License for more details.
  
   You should have received a copy of the GNU Lesser General Public
   License along with this library; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
  */
  package com.gridsystems.nextgrid.api.pom;
  
  import java.net.URI;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.PriorityQueue;
  import java.util.UUID;
  import java.util.Map.Entry;
  
  import javax.xml.namespace.QName;
  
  import nextgrid.api.env.ProcessEnvironment;
  import nextgrid.api.pem.POMListener;
  import nextgrid.api.pom.AbstractProcess;
  import nextgrid.api.pom.ControlProcess;
  import nextgrid.api.pom.Process;
  import nextgrid.api.pom.ProcessController;
  import nextgrid.api.pom.ProcessException;
  import nextgrid.api.pom.Reference;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import com.gridsystems.nextgrid.api.pom.ref.IncompatibleReferenceException;
  import com.gridsystems.nextgrid.api.pom.ref.UnresolvedReferenceException;
  
  /**
   * Base implementation for Processes.
   * <p>
   * To enact a process, the programmer should invoke the following operation:
   *
   * <pre>
   * EnactorContext context = ...;
   * Process p = ...;
   * try {
   *   p.enact(context);
   * } catch (Exception e) {
   *   ...
   * }
   * </pre>
   *
   * @author Rodrigo Ruiz
   */
  public abstract class ProcessImpl extends PemHelper implements Enactable {
  
    /**
     * Enaction logger.
     */
    protected static final Log ENACTOR_LOG = LogFactory.getLog("ENACTOR");
  
    /**
     * Initial state of the internal DFA.
     */
    public static final int DFA_INITIAL_STATE = 0;
  
    /**
     * Validation types.
     */
    protected static enum ValidationType {
      /**
       * This value is passed to {@link #doValidate(int)} before child processes
       * are validated.
       */
      BEFORE_CHILDREN,
      /**
       * This value is passed to {@link #doValidate(int)} after child processes
       * are validated.
       */
      AFTER_CHILDREN
    };
  
    /**
     * The id of this process.
     */
    private URI id;
  
    /**
     * The name of this process.
     */
    private String name;
 
   /**
    * Human-readable description of this process.
    */
   private String description;
 
   /**
    * Parent process.
    */
   private Process parent;
 
   /**
    * Map of input references. The keys are the reference IDs.
    */
   private Map<String, Reference<?>> inputs;
 
   /**
    * Map of output references. The keys are the reference IDs.
    */
   private Map<String, Reference<?>> outputs;
 
   /**
    * Set containing the names of the inputs that will be "locally" used
    * by this process.
    */
   private Map<String, Class<?>> usedInputs;
 
   /**
    * Set containing the names of the outputs that will be "locally" used
    * by this process.
    */
   private Map<String, Class<?>> usedOutputs;
 
   /**
    * Attribute map.
    */
   private Map<QName, String> attribs;
 
   /**
    * Process enaction state. Used to implement an internal DFA.
    */
   private int state = DFA_INITIAL_STATE;
 
   /**
    * Flag that marks if this process has already been validated.
    */
   private boolean validated;
 
   /**
    * Evaluation flag.
    */
   private boolean evaluated;
 
   /**
    * Lazy flag.
    */
   private boolean lazy;
 
   /**
    * Creates a new instance with a default empty model.
    */
   protected ProcessImpl() {
     this.id = genRandomURI();
     this.inputs = new HashMap<String, Reference<?>>();
     this.outputs = new HashMap<String, Reference<?>>();
     this.usedInputs = new HashMap<String, Class<?>>();
     this.usedOutputs = new HashMap<String, Class<?>>();
     this.attribs = new HashMap<QName, String>();
   }
 
   /**
    * {@inheritDoc}
    */
   public final Map<QName, String> getAttributes() {
     return this.attribs;
   }
 
   /**
    * {@inheritDoc}
    */
   public final String getAttribute(QName name) {
     return this.attribs.get(name);
   }
 
   /**
    * {@inheritDoc}
    */
   public final String getAttribute(String name) {
     return this.attribs.get(new QName("", name));
   }
 
   /**
    * {@inheritDoc}
    */
   public final String getAttribute(String ns, String local) {
     return this.attribs.get(new QName(ns, local));
   }
 
   /**
    * {@inheritDoc}
    */
   public final void setAttribute(QName name, String value) {
     if (value == null) {
       this.attribs.remove(name);
     } else {
       this.attribs.put(name, value);
     }
   }
 
   /**
    * {@inheritDoc}
    */
   public final void setAttribute(String name, String value) {
     final QName key = new QName("", name);
     if (value == null) {
       this.attribs.remove(key);
     } else {
       this.attribs.put(key, value);
     }
   }
 
   /**
    * Gets the id of this process instance. The id must be unique within the
    * scope this instance is defined in.
    *
    * @return The process id
    */
   public final URI getId() {
     return this.id;
   }
 
   /**
    * Gets the name of this process. This is just a human-readable label for
    * presentation purposes.
    *
    * @return The process name
    */
   public final String getName() {
     return this.name;
   }
 
   /**
    * Gets a human-readable description of this process.
    *
    * @return A human-readable description of this process
    */
   public final String getDescription() {
     return this.description;
   }
 
   /**
    * Gets the parent process, or null if there is no parent.
    *
    * @return The parent process
    */
   public final Process getParent() {
     return this.parent;
   }
 
   /**
    * Gets the internal DFA state of this process.
    *
    * @return The internal DFA state
    */
   public final int getState() {
     return this.state;
   }
 
   /**
    * Sets the internal DFA state of this process.
    *
    * @param state The internal DFA state
    */
   protected final void setState(int state) {
     this.state = state;
   }
 
   /**
    * Searches up through the process hierarchy to find the root process of the
    * workflow this process is defined in.
    *
    * @return The root process of the workflow
    */
   public final Process findRoot() {
     Process root = this;
     Process localParent = root.getParent();
     while (localParent != null) {
       root = localParent;
       localParent = root.getParent();
     }
 
     return root;
   }
 
   /**
    * Sets the id of this process instance.
    *
    * @param id A unique identifier
    * @throws NullPointerException     If id is null
    */
   public final void setId(URI id) {
     if (id == null) {
       throw new NullPointerException("Null id");
     } else {
       this.id = id;
     }
   }
 
   /**
    * Sets this process name.
    *
    * @param name The process name
    */
   public final void setName(String name) {
     this.name = name;
   }
 
   /**
    * Sets the description of this process.
    *
    * @param description A textual description
    */
   public final void setDescription(String description) {
     this.description = description;
   }
 
   /**
    * Sets the parent process instance.
    *
    * @param parent The parent process
    */
   public final void setParent(Process parent) {
     this.parent = parent;
   }
 
   /**
    * {@inheritDoc}
    */
   public final void validate() throws ProcessException {
     resolveReferences();
 
     // Local validation
     doValidate(ValidationType.BEFORE_CHILDREN);
 
     // Validate children processes
     if (this instanceof AbstractProcess) {
       Process p = ((AbstractProcess)this).getSelected();
       if (p != null) {
         p.validate();
       }
     } else if (this instanceof ControlProcess) {
       for (Process p : ((ControlProcess)this).getChildren()) {
         p.validate();
       }
     }
 
     // Local validation
     doValidate(ValidationType.AFTER_CHILDREN);
 
     // Mark this process as already validated
     validated = true;
   }
 
   /**
    * {@inheritDoc}
    */
   public final void invalidate() {
     validated = false;
   }
 
   /**
    * {@inheritDoc}
    */
   public final boolean isLazy() {
     return this.lazy;
   }
 
   /**
    * {@inheritDoc}
    */
   public final void setLazy(boolean lazy) {
     this.lazy = lazy;
   }
 
   /**
    * {@inheritDoc}
    */
   public final boolean isValidated() {
     return this.validated;
   }
 
   /**
    * {@inheritDoc}
    */
   public final boolean isEvaluated() {
     return this.evaluated;
   }
 
   /**
    * Sets the evaluation flag value.
    *
    * @param evaluated The new value
    */
   protected final void setEvaluated(boolean evaluated) {
     this.evaluated = evaluated;
   }
 
   /**
    * Shortcut for the interface method <tt>prioritise</tt>.
    *
    * @param env The process environment to use
    * @return A set of abstract processes
    * @throws ProcessException If an error occurs
    */
   protected final PriorityQueue<Process> prioritise(ProcessEnvironment env)
     throws ProcessException {
 
     PriorityQueue<Process> set = new PriorityQueue<Process>(1,
       new ProcessPriorityComparator()
     );
     this.prioritise(env, set);
 
     return set;
   }
 
   /**
    * Performs any validation action needed for the process implementation.
    *
    * @param type Specifies when this validation is being invoked
    * @throws ProcessException If the validation fails
    */
   protected abstract void doValidate(ValidationType type)
     throws ProcessException;
 
   /**
    * {@inheritDoc}
    */
   public final void evaluate(ProcessEnvironment env) throws ProcessException {
     this.evaluate(env, false);
   }
 
   /**
    * Evaluates this process. It uses the "force" parameter to override the
    * value of the lazy attribute.
    *
    * @param env   The environment to use
    * @param force Whether to force the evaluation or not
    * @throws ProcessException If an error occurs
    */
   public final void evaluate(ProcessEnvironment env, boolean force)
     throws ProcessException {
     if (!isEvaluated()) {
       ENACTOR_LOG.debug("Evaluating process...");
       attachListeners(env);
       ENACTOR_LOG.debug("Listeners attached");
 
       if (force || !lazy) {
         ENACTOR_LOG.debug("Evaluation required. Performing...");
         try {
           doEvaluate(env);
           setEvaluated(true);
           fireProcessEvaluated();
           ENACTOR_LOG.debug("Evaluation finished (" + this.getId() + ")");
         } catch (ProcessException e) {
           fireProcessFailed(e);
           throw e;
         }
       }
     }
   }
 
   /**
    * Performs the actual actions needed for evaluating this process instance.
    *
    * @param env The process environment
    * @throws ProcessException If an error occurs
    */
   public abstract void doEvaluate(ProcessEnvironment env) throws ProcessException;
 
   /**
    * {@inheritDoc}
    */
   public final ProcessController enact(ProcessEnvironment env)
     throws ProcessException {
     evaluate(env, true);
 
     if (env instanceof ProcessContext) {
       ProcessContext ctx = (ProcessContext)env;
       return new ThreadController(ctx, this);
     } else {
       throw new IllegalArgumentException("Only ProcessContext is supported");
     }
   }
 
   /**
    * Performs specific reset actions in each subclass.
    */
   protected abstract void doReset();
 
   /**
    * Resets any child process.
    */
   protected abstract void resetChildren();
 
   /**
    * {@inheritDoc}
    */
   public final void reset() {
     this.evaluated = false;
 
     for (Reference<?> ref : this.getOutputs().values()) {
       ref.setValue(null);
     }
 
     resetChildren();
 
     doReset();
   }
 
   /**
    * Attaches listeners configured in the environment to this process.
    *
    * @param env The environment
    */
   protected final void attachListeners(ProcessEnvironment env) {
     ENACTOR_LOG.debug("Looking for listeners for " + this.getId());
     Collection<POMListener> collection = env.getListenersFor(this.getId());
     for (POMListener l : collection) {
       ENACTOR_LOG.debug("Found listener for " + this.getId() + ": " + l);
       this.addListener(l);
     }
   }
 
   // -------------------------------------------------------------
   // Input/Output management
   // -------------------------------------------------------------
 
   /**
    * Gets the map of references to input parameters.
    *
    * @return the input parameter reference map
    */
   public final Map<String, Reference<?>> getInputs() {
     Map<String, Reference<?>> map = new HashMap<String, Reference<?>>();
     getInputs(map);
     return Collections.unmodifiableMap(map);
   }
 
   /**
    * Gets all inputs of this process, including inherited ones into the
    * specified map.
    *
    * @param map The map to fill
    */
   protected final void getInputs(Map<String, Reference<?>> map) {
     if (map != null) {
       if (this.getParent() instanceof ProcessImpl) {
         ((ProcessImpl)this.getParent()).getInputs(map);
       } else if (this.getParent() != null) {
         map.putAll(this.getParent().getInputs());
       }
       map.putAll(this.inputs);
     }
   }
 
   /**
    * Gets the map of references to output parameters.
    *
    * @return the output parameter reference map
    */
   public final Map<String, Reference<?>> getOutputs() {
     Map<String, Reference<?>> map = new HashMap<String, Reference<?>>();
     getOutputs(map);
     return Collections.unmodifiableMap(map);
   }
 
   /**
    * Gets all outputs of this process, including inherited ones into the
    * specified map.
    *
    * @param map The map to fill
    */
   protected final void getOutputs(Map<String, Reference<?>> map) {
     if (map != null) {
       if (this.getParent() instanceof ProcessImpl) {
         ((ProcessImpl)this.getParent()).getOutputs(map);
       } else if (this.getParent() != null) {
         map.putAll(this.getParent().getOutputs());
       }
       map.putAll(this.outputs);
     }
   }
 
   /**
    * Gets an input parameter by its name.
    * <p>
    * Parameters are inheritable, so if a local parameter is not found, the
    * search will be continued on the parent process.
    *
    * @param key The parameter name
    * @return The parameter, or null if none exists with the specified name
    */
   public final Reference<?> getInput(String key) {
     Reference<?> ref = this.inputs.get(key);
     if (ref == null  && this.getParent() != null) {
       ref = this.getParent().getInput(key);
     }
     return ref;
   }
 
   /**
    * Gets an output parameter by its name.
    * <p>
    * Parameters are inheritable, so if a local parameter is not found, the
    * search will be continued on the parent process.
    *
    * @param key The parameter name
    * @return The parameter, or null if none exists with the specified name
    */
   public final Reference<?> getOutput(String key) {
     if (this.outputs.containsKey(key)) {
       return this.outputs.get(key);
     } else if (this.getParent() != null) {
       return this.getParent().getOutput(key);
     } else {
       return null;
     }
   }
 
   /**
    * Puts a reference into the map of input parameters for this process.
    *
    * @param key   The parameter name
    * @param input The parameter
    */
   public final void putInput(String key, Reference<?> input) {
     synchronized (inputs) {
       Reference<?> old;
       if (input == null) {
         old = inputs.remove(key);
       } else {
         Class<?> type = usedInputs.get(key);
         if (type != null && !input.canCastTo(type)) {
           throw new ClassCastException("Type incompatible with resolved reference");
         }
         input.addReader(this);
         old = inputs.put(key, input);
       }
       if (old != null) {
         old.removeReader(this);
       }
     }
   }
 
   /**
    * Puts a reference into the map of output parameters for this process.
    *
    * @param key    The parameter name
    * @param output The parameter
    */
   public final void putOutput(String key, Reference<?> output) {
     synchronized (outputs) {
       Reference<?> old;
       if (output == null) {
         old = outputs.remove(key);
       } else {
         Class<?> type = usedOutputs.get(key);
         if (type != null && !output.canCastFrom(type)) {
           ENACTOR_LOG.warn("Type incompatible with resolved reference");
           ENACTOR_LOG.warn("- name: " + key);
           ENACTOR_LOG.warn("- type: " + type);
           ENACTOR_LOG.warn("- output: " + output.getValueType());
           throw new ClassCastException("Type incompatible with resolved reference");
         }
         output.addWriter(this);
         old = outputs.put(key, output);
       }
       if (old != null) {
         old.removeWriter(this);
       }
     }
   }
 
   /**
    * Removes a reference from the input map.
    *
    * @param key The name of the reference to remove
    */
   public final void removeInput(String key) {
     putInput(key, null);
   }
 
   /**
    * Removes a reference from the output map.
    *
    * @param key The name of the reference to remove
    */
   public final void removeOutput(String key) {
     putOutput(key, null);
   }
 
   /**
    * {@inheritDoc}
    */
   public final void useInput(String name, Class<?> type) {
     Reference<?> ref = getInput(name);
     if (ref != null && !ref.canCastTo(type)) {
       throw new ClassCastException("Type incompatible with resolved reference");
     }
     usedInputs.put(name, type);
   }
 
   /**
    * {@inheritDoc}
    */
   public final void useOutput(String name, Class<?> type) {
     Reference<?> ref = getOutput(name);
     if (ref != null && !ref.canCastTo(type)) {
       throw new ClassCastException("Type incompatible with resolved reference");
     }
     usedOutputs.put(name, type);
   }
 
   /**
    * {@inheritDoc}
    */
   public final void unuseInput(String... names) {
     for (String n : names) {
       usedInputs.remove(n);
     }
   }
 
   /**
    * {@inheritDoc}
    */
   public final void unuseOutput(String... names) {
     for (String n : names) {
       usedOutputs.remove(n);
     }
   }
 
   /**
    * {@inheritDoc}
    */
   public final String[] getUsedInputNames() {
     return usedInputs.keySet().toArray(new String[usedInputs.size()]);
   }
 
   /**
    * {@inheritDoc}
    */
   public final String[] getUsedOutputNames() {
     return usedOutputs.keySet().toArray(new String[usedOutputs.size()]);
   }
 
   /**
    * {@inheritDoc}
    */
   public final Class<?> getInputType(String name) {
     return usedInputs.get(name);
   }
 
   /**
    * {@inheritDoc}
    */
   public final Class<?> getOutputType(String name) {
     return usedOutputs.get(name);
   }
 
   /**
    * Searches through all processes in the workflow (considering this process
    * as the root node), and tries to resolve any used input or output to
    * a reference.
    * <p>
    * During this process, reference reader and writer lists are synchronised
    * with the current workflow layout.
    *
    * @throws ProcessException If a reference cannot be resolved or an
    *                          incompatibility is found
    */
   private void resolveReferences() throws ProcessException {
 
     for (Entry<String, Class<?>> entry : usedInputs.entrySet()) {
       final String key = entry.getKey();
       final Class<?> type = entry.getValue();
 
       Reference<?> ref = getInput(key);
       if (ref == null) {
         throw new UnresolvedReferenceException(key + "@" + this.getName());
       } else if (ref.canCastTo(type)) {
         ref.addReader(this);
       } else {
         throw new IncompatibleReferenceException(key);
       }
     }
 
     for (Entry<String, Class<?>> entry : usedOutputs.entrySet()) {
       final String key = entry.getKey();
       final Class<?> type = entry.getValue();
 
       Reference<?> ref = getOutput(key);
       if (ref == null) {
         throw new UnresolvedReferenceException(key);
       } else if (ref.canCastTo(type)) {
         ref.addReader(this);
       } else {
         throw new IncompatibleReferenceException(key);
       }
     }
   }
 
   // -------------------------------------------------------------
   // Miscelanea
   // -------------------------------------------------------------
 
   /**
    * Waits until all input parameters are available for reading.
    *
    * @throws InterruptedException If the thread is interrupted
    */
   protected void waitForInputs() throws InterruptedException {
     ENACTOR_LOG.debug("Process '" + this.getName() + "': waiting for inputs");
     for (Reference<?> ref : inputs.values()) {
       ref.waitAvailable();
     }
   }
 
   /**
    * Generates a random URI in the UUID name space.
    *
    * @return An UUID encapsulated within a URI
    */
   private URI genRandomURI() {
     UUID uuid = UUID.randomUUID();
     try {
       return new URI("urn:uuid:" + uuid);
     } catch (Exception e) {
       throw new InternalError(e.toString());
     }
   }
 
   /**
    * {@inheritDoc}
    */
   @Override public Process copy() {
     ProcessImpl copy = (ProcessImpl)super.copy();
     copy.attribs = new HashMap<QName, String>(this.attribs);
     copy.inputs = copyRefs(this.inputs);
     copy.outputs = copyRefs(this.outputs);
     copy.usedInputs = copyTypes(this.usedInputs);
     copy.usedOutputs = copyTypes(this.usedOutputs);
 
     copy.parent = null;
 
     return copy;
   }
 
   /**
    * Creates a copy of the specified reference map.
    *
    * @param map The map to copy
    * @return The copy
    */
   private Map<String, Reference<?>> copyRefs(Map<String, Reference<?>> map) {
     Map<String, Reference<?>> copy = new HashMap<String, Reference<?>>();
 
     for (Map.Entry<String, Reference<?>> entry : map.entrySet()) {
       copy.put(entry.getKey(), entry.getValue().copy());
     }
 
     return copy;
   }
 
   /**
    * Creates a copy of the specified parameter type map.
    *
    * @param map The map to copy
    * @return The copy
    */
   private Map<String, Class<?>> copyTypes(Map<String, Class<?>> map) {
     Map<String, Class<?>> copy = new HashMap<String, Class<?>>();
 
     for (Map.Entry<String, Class<?>> entry : map.entrySet()) {
       copy.put(entry.getKey(), entry.getValue());
     }
 
     return copy;
   }
 }