/*

  * Copyright 2007, 2008 Ange Optimization ApS

  *

  * Licensed under the Apache License, Version 2.0 (the "License");

  * you may not use this file excep 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.

  */

 /**

  * @author Kim Hansen

  */

 package eu.simuline.octave.exec;

 import java.io.BufferedReader;

 import java.io.File;

 import java.io.IOException;

 import java.io.InputStreamReader;

 import java.io.OutputStreamWriter;

 import java.io.Writer;

 import java.util.Random;

 import java.util.concurrent.CancellationException;

 import java.util.concurrent.ExecutionException;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.Future;

 import java.util.concurrent.ThreadFactory;

 import eu.simuline.octave.exception.OctaveException;

 import eu.simuline.octave.exception.OctaveIOException;

 import eu.simuline.octave.util.NamedThreadFactory;

 import eu.simuline.octave.util.NoCloseWriter;

 import eu.simuline.octave.util.ReaderWriterPipeThread;

 import eu.simuline.octave.util.TeeWriter;

 import eu.simuline.octave.OctaveUtils;

 import org.apache.commons.logging.Log;

 import org.apache.commons.logging.LogFactory;

 /**

  * The object connecting to the octave process. 

  */

 public final class OctaveExec {

     public static final String MSG_IOE_NH = 

         "InterruptedException should not happen";

     public static final String MSG_EXE_NH = 

         "ExecutionException should not happen";

     public static final String MSG_RTE_NH = 

         "RuntimeException should not happen";

     private static final Log LOG = LogFactory.getLog(OctaveExec.class);

     /**

      * The octave process created in the constructor 

      * with given command, arguments, environment and working directory. 

      */

     private final Process process;

     /**

      * 

      */

     private final Writer processWriter;

     /**

      * The input reader for {@link #process}. 

      * This is used by {@link #evalRW(WriteFunctor, ReadFunctor)} 

      * and used to close via {@link #close()}. 

      */

     private final BufferedReader processReader;

     /**

      * Used in method {@link #evalRW(WriteFunctor, ReadFunctor)} 

      * to submit essentially the write functor which submits the input 

      * and thereafter the read function which collects the output. 

      * Besides this, the executor is invoked to shutdown. 

      */

     private final ExecutorService executor;

     /**

      * The error thread of the error stream of {@link #process} 

      * writing the error stream to a given writer. 

      * This is used to close but also to change the error writer 

      * by {@link #setErrorWriter(Writer)}. 

      */

     private final ReaderWriterPipeThread errorStreamThread;

     private boolean destroyed = false;

     /**

      * Will start the octave process.

      *

      * @param numThreadsReuse

      *    the number of threads to be reused in a fixed thread pool. 

      *    This is either positive or <code>-1</code>, 

      *    which means that a cached thread pool is used instead of a fixed one. 

      * @param stdinLog

      *    This writer will capture all

      *    that is written to the octave process via stdin,

      *    if null the data will not be captured.

      * @param stderrLog

      *    This writer will capture all

      *    that is written from the octave process on stderr,

      *    if null the data will not be captured. 

      * @param cmdArray

      *    The array consisting of command and arguments: 

      *    The 0th entry is either the path to the octave program,

      *    or the command found by looking at the builtin variable "paths" 

      *    reconstructing the path. 

      *    starting with the 1th entry, 

      *    may follow the array of arguments to start the octave program with. 

      *    CAUTION: allowed values depend on the octave version. 

      * @param environment

      *    The environment for the octave process, 

      *    i.e. the set of values of environment variables 

      *    if null the process will inherit the environment

      *    for the virtual machine. 

      *    If not null, each entry has the form <code>name=value</code>. 

      * @param workingDir

      *    This will be the working dir for the octave process,

      *    if null the process will inherit the working dir

      *    of the current process.

      */

     public OctaveExec(final int numThreadsReuse,

                       final Writer stdinLog, 

                       final Writer stderrLog, 

                       final String[] cmdArray,

                       final String[] environment, // always invoked with null 

                       final File workingDir) {

         ThreadFactory threadFactory = new NamedThreadFactory();

         this.executor = numThreadsReuse == -1

             ? Executors.newCachedThreadPool(threadFactory)

             : Executors.newFixedThreadPool(numThreadsReuse, threadFactory);

         try {

             this.process = Runtime.getRuntime().exec(cmdArray, 

                                                      environment, 

                                                      workingDir);

         } catch (final IOException e) {

             throw new OctaveIOException(e);

         }

         // Connect stderr

         this.errorStreamThread = ReaderWriterPipeThread

             .instantiate(new InputStreamReader(this.process.getErrorStream(), 

                                                OctaveUtils.getUTF8()),

                          stderrLog);

         // Connect stdout

         this.processReader = new BufferedReader

             (new InputStreamReader(this.process.getInputStream(), 

                                    OctaveUtils.getUTF8()));

         // Connect stdin

         Writer pw = new OutputStreamWriter(this.process.getOutputStream(),

                                            OctaveUtils.getUTF8());

         // all written to processWriter will go to pw and, 

         // if not null to stdinLog

         this.processWriter = (stdinLog == null)

             ? pw

             : new TeeWriter(new NoCloseWriter(stdinLog), pw);

     }

     private final Random random = new Random();

     private String generateSpacer() {

         return "-=+X+=- Octave.java spacer -=+X+=- " + 

             random.nextLong() + " -=+X+=-";

     }

     /**

      * Passes <code>input</code> to octave 

      * and get back <code>output</code>. 

      *

      * @param input

      *    a write functor which represents the script 

      *    to be executed in octave. 

      * @param output

      *    the read functor which reads the result of octave execution. 

      *    After evaluation of this method, 

      *    the <code>output</code> is asked for the result. 

      *///<code></code>

     // used in OctaveIO#set(Map), OctaveIO#get(String), 

     // OctaveIO#checkIfVarExists(String) and in 

     // OctaveEngine#unsafeEval(String) OctaveEngine#unsafeEval(Reader) and 

     // OctaveEngine#getVersion() only 

     public void evalRW(final WriteFunctor input, final ReadFunctor output) {

         final String spacer = generateSpacer();

         final Future<Void> writerFuture = 

             this.executor.submit(new OctaveWriterCallable(this.processWriter, 

                                                           input, 

                                                           spacer));

         final Future<Void> readerFuture = 

             this.executor.submit(new OctaveReaderCallable(this.processReader, 

                                                           output, 

                                                           spacer));

         final RuntimeException writerException = getFromFuture(writerFuture);

         // if (writerException instanceof CancellationException) {

         //     LOG.error("Did not expect writer to be canceled", 

         //               writerException);

         // }

         if (writerException != null) {

             if (writerException instanceof CancellationException) {

                 LOG.error("Did not expect writer to be canceled", 

                               writerException);

             }

             readerFuture.cancel(true); // may interrupt if running 

             throw writerException;

         }

         final RuntimeException readerException = getFromFuture(readerFuture);

         // if (writerException != null) {

         //     throw writerException;

         // }

         if (readerException != null) {

             // Only gets here when writerException==null, 

             // and in that case we don't expect the reader to be cancelled

             if (readerException instanceof CancellationException) {

                 LOG.error("Did not expect reader to be canceled", 

                           readerException);

             }

             throw readerException;

         }

     }

     /**

      * Completes computation on future 

      * and returns an exception thrown or null. 

      */

     private RuntimeException getFromFuture(Future<Void> future) {

         try {

             future.get();

         } catch (final InterruptedException e) {

             LOG.error(MSG_IOE_NH, e);

             return new RuntimeException(MSG_IOE_NH, e);

         } catch (final ExecutionException e) {

             if (e.getCause() instanceof OctaveException) {

                 final OctaveException oe = (OctaveException) e.getCause();

                 return reInstException(oe);

             }

             // Can happen when there is an error in a OctaveWriter

              LOG.error(MSG_EXE_NH, e);

             return new RuntimeException(MSG_EXE_NH, e);

         } catch (final CancellationException e) {

             return e;

         } catch (final RuntimeException e) { // NOPMD 

             LOG.error(MSG_RTE_NH, e);

             return new RuntimeException(MSG_RTE_NH, e);

         }

         return null;

     }

     /**

      * Used by {@link #getFromFuture(Future)} 

      * to reinstantiate an {@link OctaveException} 

      * if this occurs as the cause of an {@link ExecutionException}. 

      */

     private OctaveException reInstException(OctaveException exc) {

         OctaveException res;

         try {

             res = exc.getClass()

                 // may throw NoSuchMethodException 

                 // isa ReflectiveOperationException, 

                 // SecurityException isa RuntimeException 

                 .getConstructor(String.class, Throwable.class)

                 // may throw 

                 // IllegalArgumentException, 

                 // ReflectiveOperationException: 

                 // - IllegalAccessException: construtor inaccessible 

                 // - InstantiationException: Exception class is abstract 

                 // - InvocationTargetException: if the constructor throws an exc

                 // ExceptionInInitializerError

                 .newInstance(exc.getMessage(), exc);

         } catch (RuntimeException e) { // NOPMD

             throw new IllegalStateException("Exception should not happen", e);

         } catch (ReflectiveOperationException e) {

             throw new IllegalStateException("Exception should not happen", e);

         } catch (ExceptionInInitializerError e) {

             throw new IllegalStateException("Error should not happen", e);

         }

         if (isDestroyed()) {

             res.setDestroyed(true);

         }

         return res;

     }

     /**

      * Sets {@link #destroyed} to the parameter value given. 

      */

     private synchronized void setDestroyed(final boolean destroyed) {

         this.destroyed = destroyed;

     }

     /**

      * Returns {@link #destroyed}. 

      */

     private synchronized boolean isDestroyed() {

         return this.destroyed;

     }

     /**

      * Kill the octave process without remorse. 

      */

     public void destroy() {

         setDestroyed(true);

         this.executor.shutdownNow(); // returns list of tasks awaiting exec. 

         this.process.destroy();

         this.errorStreamThread.close();

         try {

             this.processWriter.close();

         } catch (final IOException e) {

             LOG.debug("Ignored error from processWriter.close() " + 

                       "in OctaveExec.destroy()", e);

         }

     }

     /**

      * Close the octave process in an orderly fashion.

      */

     public void close() {

         try {

             // it is not worth it to rewrite this 

             // to use eval() and some specialised Functors

             this.processWriter.write("exit\n");

             this.processWriter.close();

             final String read1 = this.processReader.readLine();

             // Allow a single blank line, exit in octave 3.2 returns that:

             if (read1 != null && !"".equals(read1)) {

                 throw new OctaveIOException

                     ("Expected a blank line, read '" + read1 + "'");

             }

             final String read2 = this.processReader.readLine();

             if (read2 != null) {

                 throw new OctaveIOException

                     ("Expected reader to be at end of stream, read '" + 

                      read2 + "'");

             }

             this.processReader.close();

             this.errorStreamThread.close();

             int exitValue;

             try {

                 exitValue = this.process.waitFor();

             } catch (final InterruptedException e) {

                 throw new OctaveIOException

                     ("Interrupted when waiting for octave process " + 

                      "to terminate", e);

             }

             if (exitValue != 0) {

                 throw new OctaveIOException

                     ("octave process terminated abnormaly, exitValue='" + 

                      exitValue + "'");

             }

         } catch (final IOException e) {

             final OctaveIOException octaveException = 

                 new OctaveIOException("reader error", e);

             if (isDestroyed()) {

                 octaveException.setDestroyed(true);

             }

             throw octaveException;

         } finally {

             this.executor.shutdown();

         }

     }

     /**

      * @param writer

      *    the new writer to write the error output to

      */

     public void setErrorWriter(final Writer writer) {

         this.errorStreamThread.setWriter(writer);

     }

 }