Coverage Report

 /*
  * #%L
  * JavaHg
  * %%
  * Copyright (C) 2011 aragost Trifork ag
  * %%
  * Permission is hereby granted, free of charge, to any person obtaining a copy
  * of this software and associated documentation files (the "Software"), to deal
  * in the Software without restriction, including without limitation the rights
  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  * copies of the Software, and to permit persons to whom the Software is
  * furnished to do so, subject to the following conditions:
  * 
  * The above copyright notice and this permission notice shall be included in
  * all copies or substantial portions of the Software.
  * 
  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  * THE SOFTWARE.
  * #L%
  */
 package com.aragost.javahg.internals;
 
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;
 import java.nio.charset.CharacterCodingException;
 import java.nio.charset.Charset;
 import java.nio.charset.CharsetDecoder;
 import java.nio.charset.CharsetEncoder;
 import java.nio.charset.CodingErrorAction;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
 import com.aragost.javahg.MercurialExtension;
 import com.aragost.javahg.log.Logger;
 import com.aragost.javahg.log.LoggerFactory;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.google.common.io.Closeables;
 
 /**
  * Java class representing a Mercurial commandserver
  */
 public class Server {
 
     private static final Logger LOG = LoggerFactory.getLogger(Server.class);
 
     /**
      * Enum to represent the state of the server.
      * <p>
      * The state will always go from lower ordinal to higher
      */
     private enum State {
         NOT_STARTED, STARTING, RUNNING, STOPPING, STOPPED, CRASHED
     };
 
     private static final byte[] RUNCOMMAND = "runcommand\n".getBytes();
 
     /**
      * The character encoding used for the server. For now it is
      * always utf-8, in the future there might be an API to set this.
      * Future: move to ServerPool?
      */
     private final Charset encoding;
 
     /**
      * Policy for handling decoding errors.
      * Future: move to ServerPool?
      */
     private CodingErrorAction errorAction = CodingErrorAction.REPORT;
 
     /**
      * Size of buffer to buffer stderr from Mercurial server process
      */
     private int stderrBufferSize = 1024;
 
     /**
      * Stderr from Mercurial server process generated duing start up.
      */
     private String startupStderr = "";
 
     /**
      * The underlying OS process
      */
     private Process process;
 
     /**
      * Thread that read stderr from command server
      * <p>
      * In general if the server writes something to stderr (not the
      * 'e' channel but the actual stderr) it is something seriously
      * wrong and the server will me stopped and an exception thrown.
      * During start up of the server messages to stdout is accepted.
      */
     private StderrReader errorReaderThread;
 
     private State state = State.NOT_STARTED;
 
     /**
      * If non-null the {@link AbstractCommand} the server is currently
      * executing.
      * <p>
      * The server is single treaded and can only execute one command
      * at a time.
      */
     private AbstractCommand currentCommand;
 
     /**
      * If the appropriate log level is enabled and {@link #currentCommand} is
      * not null then this contains a log message.
      */
     private String currentLog;
 
     /**
      * If the appropriate log level is enabled and {@link #currentCommand} is
      * not null then this contains the start time.
      */
     private long currentStartTime;
 
     /**
      * The directory containing the Mercurial repository where the
      * server is running.
      */
     private File directory;
 
     /**
      * Location of the Mercurial binary.
      */
     private final String hgBin;
 
     /**
      * Time this server was last active.
      */
     private volatile long lastActiveTime;
 
     /**
      * Enable access to pending changesets.
      */
     private boolean enablePendingChangesets = false;
 
     /**
      * Create a new Server object
      * 
      * @param hgBin
      *            the Mercurial binary to use
      */
     public Server(String hgBin, Charset encoding) {
         this.hgBin = hgBin;
         this.encoding = encoding;
     }
 
     public CodingErrorAction getErrorAction() {
         return errorAction;
     }
 
     public void setErrorAction(CodingErrorAction errorAction) {
         this.errorAction = errorAction;
     }
 
     public int getStderrBufferSize() {
         return stderrBufferSize;
     }
 
     public String getStartupStderr() {
         return startupStderr;
     }
 
     public void setEnablePendingChangesets(boolean enablePendingChangesets){
       this.enablePendingChangesets = enablePendingChangesets;
     }
 
     public boolean isEnablePendingChangesets(){
       return enablePendingChangesets;
     }
 
     /**
      * Set the buffer size for stderr from Mercurial server process.
      * <p>
      * There is probably no reason to set this, but it is here to
      * facilitate testcases for buffer overflow.
      * 
      * @param stderrBufferSize
      */
     public void setStderrBufferSize(int stderrBufferSize) {
         this.stderrBufferSize = stderrBufferSize;
     }
 
     /**
      * @return a new {@link CharsetDecoder} instance for this server.
      */
     public CharsetDecoder newDecoder() {
         CharsetDecoder decoder = this.encoding.newDecoder();
         decoder.onMalformedInput(this.errorAction);
         decoder.onUnmappableCharacter(this.errorAction);
         return decoder;
     }
 
     /**
      * @return a new {@link CharsetEncoder} instance for this server.
      */
     public CharsetEncoder newEncoder() {
         CharsetEncoder encoder = this.encoding.newEncoder();
         encoder.onMalformedInput(this.errorAction);
         encoder.onUnmappableCharacter(this.errorAction);
         return encoder;
     }
 
     /**
      * Start the server in the specified directory. The directory is must be the
      * root directory of a Mercurial repository.
      * 
      * @param directory
      *            The directory of to start the command server in
      * @param hgrcPath
      *            The path to the hgrc config file to use. May be null
      * @param extraArguments
      *            Additional argument to start the command server with. Eg
      *            extensions to enable
      * @param env Optional map of environment variables to set when starting 
      *            the command server. 
      * @param supervisor
      *            Optional function to be executed periodically by the error
      *            stream thread. May be null
      * @return The hello message from the server
      */
     public String start(File directory, final String hgrcPath, List<String> extraArguments, Map<String, String> env, Runnable supervisor) {
         this.directory = directory.getAbsoluteFile();
         if (!new File(directory, ".hg").isDirectory()) {
             throw new IllegalArgumentException("No .hg in " + directory);
         }
         String result;
         try {
             List<String> args = Lists.newArrayList("serve", "--cmdserver", "pipe", "--config", "ui.interactive=true",
                     "--config", "ui.merge=internal:fail");
             ArrayList<Class<? extends MercurialExtension>> extList = Lists.newArrayList();
             extList.add(JavaHgMercurialExtension.class);
             args.addAll(ExtensionManager.getInstance().process(extList));
             args.addAll(extraArguments);
             this.state = State.STARTING;
             this.process = execHg(this.directory, hgrcPath, args, env);
 
             active();
             this.errorReaderThread = new StderrReader(this.process, this.stderrBufferSize, supervisor);
             this.errorReaderThread.start();
             // If for some reason the server failed to start the
             // creation of the BlockInputStream will fail (because the
             // the stdout stream on the process has been closed). So
             // before we try to create the BlockInputStream the error
             // reader thread should be started to print the error
             BlockInputStream block = new BlockInputStream(this.process.getInputStream());
             result = Utils.readStream(block, newDecoder());
             // The hello message from the server has been read and it
             // is now running normally.
 
             checkStderr();
             this.state = State.RUNNING;
 
             LOG.info("Command server started: {}", this.directory);
         } catch (Exception e) {
             verifyServerProcess(e);
             throw Utils.asRuntime(e);
         }
         verifyServerProcess(null);
         return result;
     }
 
     private static final int ERROR_READER_THREAD_TIMEOUT = 5000; // milliseconds
 
     /**
      * Stop the Mercurial server process
      */
     void stop() {
         if (this.process == null || this.state == State.STOPPED) {
             LOG.warn("Trying to stop already stopped server");
             return;
         }
 
         this.state = State.STOPPING;
         Closeables.closeQuietly(this.process.getOutputStream());
         Closeables.closeQuietly(this.process.getInputStream());
         this.errorReaderThread.finish();
         this.errorReaderThread = null;
         Closeables.closeQuietly(this.process.getErrorStream());
         // Closing the streams of the process will stop the
         // process and close the error stream, which will return int
         // the errorReaderThread to terminate.
         try {
             this.process.waitFor();
         } catch (InterruptedException e) {
             LOG.error("Process for Mercurial server interrupted", e);
             throw Utils.asRuntime(e);
         }
         LOG.info("Command server stopped: {}", this.directory);
         this.currentCommand = null;
         this.process = null;
         this.directory = null;
     }
 
     private void checkStderr() {
         String s = this.errorReaderThread.bufferAsString(newDecoder());
         if (s.length() > 0) {
             LOG.error("stderr from Mercurial: {}", s);
             switch (this.state) {
             case STARTING:
                 this.startupStderr += s;
                 break;
             case STOPPING:
             case CRASHED:
                 // stderr is accepted, and logged as error
                 break;
             case RUNNING:
                 // stderr is not accepted, stop server and throw
                 // exception
                 stop();
                 // TODO What exception would be appropiate to throw?
                 throw new RuntimeException(s);
             default:
                 throw new RuntimeException("This should not happen");
             }
         }
     }
 
     /**
      * Verify that the server process hasn't terminated. If it has
      * throw an {@link UnexpectedServerTerminationException},
      * otherwise just return normally.
      * 
      * @param exception
      */
     void verifyServerProcess(Exception exception) {
         if (exception instanceof UnexpectedServerTerminationException) {
             // Already the right type
             throw (UnexpectedServerTerminationException) exception;
         }
         if (this.process != null) {
             // When the process shutdowns there is a small window
             // where the input stream is closed but the
             // process is still running. If the input stream is
             // closed we get a
             // BlockInputStream.InvalidStreamException
             // exception. In this case loop a few times with a delay
             // do see if the process did actually exit.
             // Similar the is a small window when writing to the
             // output stream, here we get an IOException
             int n = 0;
             if (exception instanceof RuntimeIOException) {
                 exception = ((RuntimeIOException) exception).getIOException();
             }
             while (true) {
                 try {
                     int exitValue = this.process.exitValue();
                     this.state = State.CRASHED;
                     checkStderr();
                     String msg = "Server process terminated premature with: " + exitValue;
                     System.err.println("JavaHg: " + msg);
                     this.process = null;
                     throw new UnexpectedServerTerminationException(exitValue, exception);
                 } catch (IllegalThreadStateException e) {
                     if (exception instanceof BlockInputStream.InvalidStreamException
                             || exception instanceof IOException) {
                         if (n++ == 4) {
                             return;
                         }
                         sleep(100);
                     } else {
                         return;
                     }
 
                 }
             }
         }
     }
 
     /**
      * Run the specified command and return a stream with the content
      * of the output channel.
      * <p>
      * The client <em>must</em> empty the return stream. The server
      * will not accept other commands until all out is read.
      * 
      * @param cmdLine
      * @param command
      * @return the standard output from the command.
      * @throws IOException
      */
     public OutputChannelInputStream runCommand(List<String> cmdLine, AbstractCommand command) throws IOException {
         if (this.currentCommand != null) {
             throw new IllegalStateException("Trying to execute new command when command already running: "
                     + this.currentCommand);
         }
 
         if (LOG.isInfoEnabled()) {
             StringBuilder buf = new StringBuilder(256);
 
             for (String s : cmdLine) {
                 buf.append(Utils.obfuscateLoginData(s));
                 buf.append(' ');
             }
 
             currentLog = buf.toString(); 
             currentStartTime = System.currentTimeMillis(); 
         }
         
         this.currentCommand = command;
         sendCommand(cmdLine);
         OutputChannelInputStream stdout = new OutputChannelInputStream(this.process.getInputStream(), this, command);
         checkStderr();
         return stdout;
     }
 
     void clearCurrentCommand(AbstractCommand cmd) {
         if (cmd != this.currentCommand) {
             throw new IllegalStateException("Wrong command");
         }
         active();
         checkStderr();
         this.currentCommand = null;
 
         if (LOG.isInfoEnabled()) {
             LOG.info("runcommand({}ms) {}", System.currentTimeMillis() - currentStartTime,
                     currentLog);
         }
     }
 
     private ByteArrayOutputStream baos = new ByteArrayOutputStream();
 
     private void sendCommand(List<String> cmdLine) throws IOException {
         active();
         baos.reset();
         CharsetEncoder encoder = newEncoder();
         encode(cmdLine.get(0), baos, encoder);
         for (String s : cmdLine.subList(1, cmdLine.size())) {
             baos.write('\0');
             encode(s, baos, encoder);
         }
 
         OutputStream outputStream = this.process.getOutputStream();
         try {
             outputStream.write(RUNCOMMAND);
             Utils.writeBigEndian(baos.size(), outputStream);
             baos.writeTo(outputStream);
             outputStream.flush();
         } catch (IOException e) {
             verifyServerProcess(e);
             throw e;
         }
     }
 
     private Process execHg(File directory, String hgrcPath, List<String> arguments, Map<String, String> env) throws IOException {
         ArrayList<String> cmdLine = Lists.newArrayList(this.hgBin);
         cmdLine.addAll(arguments);
         ProcessBuilder processBuilder = new ProcessBuilder(cmdLine);
         if (directory != null) {
             processBuilder.directory(directory);
         }
         Map<String, String> environment = processBuilder.environment();
 
         if (env != null)
         {
             for (Iterator<String> it = env.keySet().iterator(); it.hasNext();)
             {
                 String key = (String)it.next();
 
                 environment.put(key, env.get(key));
             }
         }
 
         environment.put("HGENCODING", this.encoding.displayName());
         environment.put("HGPLAIN", "1");
         if (hgrcPath != null) {
             environment.put("HGRCPATH", hgrcPath);
         }
         if (enablePendingChangesets){
           environment.put("HG_PENDING", directory.getAbsolutePath());
         }
         return processBuilder.start();
     }
 
     /**
      * Convenience method to initialize a mercurial repository in a
      * directory.
      * <p>
      * This method is not using any commandserver functionality
      * 
      * @param directory
      */
     public void initMecurialRepository(File directory) {
         execHgCommand(null, "", "init", directory.getAbsolutePath());
     }
 
     /**
      * Convenience method to clone a mercurial repository in a directory.
      * <p>
      * This method is not using any commandserver functionality
      *
      * @param directory
      * @param hgrcPath
      * @param cloneUrl
      */
     public void cloneMercurialRepository(File directory, String hgrcPath, String cloneUrl) {
       execHgCommand(null, hgrcPath, "clone", cloneUrl, directory.getAbsolutePath());
     }
 
     private void execHgCommand(File directory, String hgrcPath, String... args) {
         try {
             Process process = execHg(directory, hgrcPath, Arrays.asList(args), null);
             String stderr = Utils.readStream(process.getErrorStream(), newDecoder());
             Utils.consumeAll(process.getInputStream());
 
             if (process.waitFor() != 0) {
                 throw new RuntimeException(stderr);
             }
         } catch (IOException e) {
             throw new RuntimeIOException(e);
         } catch (InterruptedException e) {
             throw Utils.asRuntime(e);
         }
     }
 
     @Override
     public String toString() {
         return "cmdserver@" + this.directory;
     }
 
     public void sendLine(String answer) {
         OutputStream outputStream = this.process.getOutputStream();
         try {
             byte[] bytes = (answer + "\n").getBytes(this.encoding.name());
             Utils.writeBigEndian(bytes.length, outputStream);
             outputStream.write(bytes);
             outputStream.flush();
         } catch (IOException e) {
             throw new RuntimeIOException(e);
         }
     }
 
     /**
      * Encode the string and write it to the OutputStream
      * 
      * @param s
      * @param output
      * @throws IOException
      */
     private void encode(String s, OutputStream output, CharsetEncoder encoder) throws IOException {
         ByteBuffer byteBuffer = encoder.encode(CharBuffer.wrap(s));
         output.write(byteBuffer.array(), byteBuffer.arrayOffset(), byteBuffer.limit());
     }
 
     /**
      * @return The last time this server started or finished executing a command
      */
     long getLastActiveTime() {
         return lastActiveTime;
     }
 
     /**
      * Call when this server is active
      */
     private void active() {
         lastActiveTime = System.currentTimeMillis();
     }
 
     private static void sleep(int ms) {
         try {
             Thread.sleep(ms);
         } catch (InterruptedException e) {
             throw Utils.asRuntime(e);
         }
     }
 
     @Override
     protected void finalize() {
         if (this.process != null) {
             LOG.error("Stopping cmdserver via finalize. Please explicit stop server.");
             stop();
         }
     }
 
     @VisibleForTesting
     Process getProcess() {
         return process;
     }
 
     /**
      * Thread to read stderr from the Mercurial server process. The
      * stderr is written to a {@link ByteBuffer} that is then process
      * from the main thread.
      * <p>
      * If the buffer is full before being processed then the extra
      * data from stderr is discarded silently.
      */
     static class StderrReader extends Thread {
 
         private static final Logger LOG = LoggerFactory.getLogger(StderrReader.class);
 
         private final InputStream errorStream;
 
         private final ByteBuffer stderrBuffer;
 
         private volatile boolean stop = false;
 
         private final Runnable supervisor;
 
         StderrReader(Process process, int bufSize, Runnable supervisor) {
             super("JavaHg stderr reader");
             this.errorStream = process.getErrorStream();
             this.stderrBuffer = ByteBuffer.allocate(bufSize);
             this.supervisor = supervisor;
             setDaemon(true);
         }
 
         /**
          * Finish this thread and read all available stderr data.
          */
         void finish() {
             this.stop = true;
             interrupt();
             try {
                 readAllAvailableFromStderr();
             } catch (IOException e) {
                 handleIOException(e);
             }
             try {
                 join(ERROR_READER_THREAD_TIMEOUT);
                 if (isAlive()) {
                     assert false;
                     throw new RuntimeException("thread is alive. This should not happen");
                 }
             } catch (InterruptedException e) {
                 assert false;
                 throw new RuntimeException("This should not happen");
             }
 
         }
 
         /**
          * Reset the buffer and return the content as a String
          * 
          * @param decoder
          *            used to decode the bytes as a String.
          * @return the bytes converted to a String
          */
         String bufferAsString(CharsetDecoder decoder) {
             try {
                 readAllAvailableFromStderr();
             } catch (IOException e) {
                 LOG.warn("Exception trying to read stderr: {}", e.getMessage());
                 return "";
             }
 
             ByteBuffer buf = this.stderrBuffer;
             synchronized (buf) {
                 if (buf.position() > 0) {
                     CharBuffer charBuffer;
                     buf.limit(buf.position());
                     buf.position(0);
                     try {
                         charBuffer = decoder.decode(buf);
                     } catch (CharacterCodingException e) {
                         throw Utils.asRuntime(e);
                     }
                     buf.limit(buf.capacity());
                     String s = new String(charBuffer.array(), charBuffer.arrayOffset(), charBuffer.limit());
                     buf.position(0);
                     return s;
                 } else {
                     return "";
                 }
             }
         }
 
         public void run() {
             try {
                 while (!this.stop) {
                     readAllAvailableFromStderr();
                     try {
                         Thread.sleep(1000);
                     } catch (InterruptedException e) {
                         // When stopping server the stderr thread is
                         // interrupted
                     }
                     if (supervisor != null) {
                         supervisor.run();
                     }
                 }
             } catch (IOException e) {
                 handleIOException(e);
             }
         }
 
         private void handleIOException(IOException e) {
             String message = e.getMessage();
             if (message.equals("Bad file descriptor") || message.equals("Stream Closed") || message.equals("Stream closed")) {
                 LOG.warn("errorReaderThread could not read stderr. Most likely the Mercurial server process is dead.");
             } else {
                 throw new RuntimeIOException(e);
             }
         }
 
         private void readAllAvailableFromStderr() throws IOException {
             while (this.errorStream.available() > 0) {
                 int b = this.errorStream.read();
                 if (b == -1) {
                     // The stream is at eof, the process must
                     // have exited, so stop the thread.
                     break;
                 }
                 synchronized (this.stderrBuffer) {
                     try {
                         this.stderrBuffer.put((byte) b);
                     } catch (BufferOverflowException e) {
                         // just ignore
                     }
                 }
             }
 
         }
 
     }
 }