package com.aragost.javahg.internals;
   
   import java.io.File;
   import java.nio.charset.Charset;
   import java.nio.charset.CharsetDecoder;
   import java.nio.charset.CodingErrorAction;
   import java.util.List;
   import java.util.concurrent.BlockingQueue;
   import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  
  import com.aragost.javahg.HgVersion;
  import com.aragost.javahg.Repository;
  import com.aragost.javahg.RepositoryConfiguration;
  import com.aragost.javahg.commands.VersionCommand;
  import com.aragost.javahg.internals.AbstractCommand.State;
  import com.aragost.javahg.log.Logger;
  import com.aragost.javahg.log.LoggerFactory;
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.collect.Lists;
  
  /**
   * A pool of Server instances. Use {@link #take(AbstractCommand)} and
   * {@link #put(Server)}.
   * 
   * Contains up to {@link #maxServers} servers. When the maximum number of
   * servers are running commands are queued and queued commands may be cancelled.
   */
  public class ServerPool {
  
      private static final Logger LOG = LoggerFactory.getLogger(ServerPool.class);
  
      /**
       * Period at which to check if a command waiting for a server has been
       * cancelled.
       */
      private static final long WAIT_CHECK_MILLIS = 500;
  
      /**
       * If a server can't be obtained with in this number of seconds a warning is
       * written to the log
       */
      private static final int WAIT_WARN_MILLIS = 10 * 1000;
  
      /**
       * The character encoding used for the server.
       */
      private final Charset encoding;
  
      /**
       * The number of {@link Repository} instances referencing this pool
       */
      private int refCount;
  
      /**
       * Used as a stack so that hot servers are used more often.
       */
      private final BlockingQueue<Server> freeServers = new LinkedBlockingQueue<Server>();
  
      /**
       * The maximum number of servers to use
       */
      private int maxServers;
  
      /**
       * All the currently running servers.
       */
      private List<Server> servers = Lists.newArrayList();
  
      /**
       * The version of the underlying Mercurial. It is lazy initialized.
       */
      private HgVersion hgVersion;
  
      private final RepositoryConfiguration configuration;
  
      /**
       * Mercurial repository directory
       */
      private final File directory;
  
      public ServerPool(RepositoryConfiguration conf, File directory,
              boolean performInit, String cloneUrl) {
          this.maxServers = Math.max(1, conf.getConcurrency());
          this.configuration = conf;
          this.directory = directory;
          this.encoding = conf.getEncoding();
  
          Server server = createServer();
          if (performInit) {
              server.initMecurialRepository(directory);
          } else if (cloneUrl != null) {
              server.cloneMercurialRepository(directory, conf.getHgrcPath(),
                      cloneUrl);
          }
          startServer(server);
          freeServers.add(server);
          servers.add(server);
      }
 
     /**
      * Increment the refCount for this server pool.
      */
     public void incrementRefCount() {
         this.refCount++;
     }
 
     /**
      * Decrement the refCount. If it reaches 0 then the server pool is stopped.
      */
     public void decrementRefCount() {
         this.refCount--;
         if (this.refCount == 0) {
             stop();
         }
     }
 
     private void stop() {
         synchronized (servers) {
             maxServers = 0;
             for (Server server : servers) {
                 server.stop();
             }
             servers.clear();
         }
     }
 
     public CharsetDecoder newDecoder() {
         CodingErrorAction errorAction = this.configuration
                 .getCodingErrorAction();
 
         CharsetDecoder decoder = this.encoding.newDecoder();
         decoder.onMalformedInput(errorAction);
         decoder.onUnmappableCharacter(errorAction);
         return decoder;
     }
 
     /**
      * Get a server. If there are fewer than {@link #maxServers} a new server is
      * started. If no servers are available the thread blocks until there is a
      * server available. Caller must call {@link #put(Server)} after command is
      * completed.
      * 
      * @return The next available server
      * @throws InterruptedException
      *             If interrupted while waiting for a server to become free or
      *             the command was cancelled.
      * @see #put(Server)
      */
     public Server take(AbstractCommand command) throws InterruptedException {
         Server server = freeServers.poll();
 
         if (server == null) {
             synchronized (servers) {
                 if (maxServers == 0) {
                     throw new IllegalStateException("Server pool is stopped");
                 }
                 if (servers.size() < maxServers) {
                     server = createServer();
                     startServer(server);
                     servers.add(server);
                 }
             }
 
             // Already at capacity, wait for a server to become free
             if (server == null) {
                 server = waitForServer(command);
             }
         }
 
         return server;
     }
 
     /**
      * Block the current thread until a server becomes available.
      * 
      * After {@link #WAIT_WARN_MILLIS} a warning logged. After
      * {@link RepositoryConfiguration#getCommandWaitTimeout()} an error is
      * logged and an exception is thrown.
      * 
      * @param command
      *            The command
      * @return Never null
      * @throws InterruptedException
      *             If the command is cancelled.
      */
     private Server waitForServer(AbstractCommand command)
             throws InterruptedException {
         boolean warned = false;
         long startedWaitingTime = System.currentTimeMillis();
         long failTimeoutMillis = configuration.getCommandWaitTimeout() * 1000l;
 
         // Check for cancellation twice per second
         // Log if waiting for too long
         while (true) {
             Server server = freeServers.poll(WAIT_CHECK_MILLIS,
                     TimeUnit.MILLISECONDS);
 
             if (command.getState() == State.CANCELING) {
                 throw new InterruptedException(
                         "Command cancelled while waiting for comand server to become available");
             }
 
             if (server != null) {
                 return server;
             }
 
             // Check for timeouts
             long elapsed = System.currentTimeMillis() - startedWaitingTime;
             if (!warned && elapsed > WAIT_WARN_MILLIS) {
                 LOG.warn("Waited " + (WAIT_WARN_MILLIS / 1000)
                         + " seconds for server lock without obtaining it");
                 warned = true;
             } else if (elapsed > failTimeoutMillis) {
                 String msg = "Did not obtain server lock after "
                         + failTimeoutMillis / 1000 + " seconds.";
                 LOG.error(msg);
                 throw new RuntimeException(msg);
             }
         }
     }
 
     /**
      * Return the server to the pool of available servers.
      * 
      * @param server
      *            The server to return
      * @see #take(AbstractCommand)
      * @see #abort(Server)
      */
     public void put(Server server) {
         Server unusedServer = freeServers.poll();
 
         if (unusedServer != null) {
             stop(unusedServer);
         }
 
         freeServers.add(server);
     }
 
     /**
      * Stop the given server because it is in an invalid state and not able to
      * service requests.
      * 
      * @param server
      *            The server to stop.
      */
     void abort(Server server) {
         try {
             LOG.info("Aborting server " + server);
             stop(server);
         } catch (Throwable t) {
             LOG.error("Additional error stopping server", t);
             assert false;
         }
     }
 
     /**
      * Stop the given server and remove it from the list of servers. Assumes not
      * present in freeServers.
      * 
      * @param server
      *            The server to stop
      */
     private void stop(Server server) {
         synchronized (servers) {
             servers.remove(server);
         }
         server.stop();
     }
 
     private void startServer(Server server) {
         List<String> extensionArgs = ExtensionManager.getInstance().process(
                 this.configuration.getExtensionClasses());
         Runnable supervisor = null;
         if (this.configuration.getServerIdleTime() != Integer.MAX_VALUE) {
             supervisor = new ServerSupervisor(server);
         }
 
         server.start(this.directory, this.configuration.getHgrcPath(),
                 extensionArgs, this.configuration.getEnvironment(), supervisor);
     }
 
     private Server createServer() {
         Server server = new Server(this.configuration.getHgBin(), encoding);
         server.setStderrBufferSize(this.configuration.getStderrBufferSize());
         server.setErrorAction(this.configuration.getCodingErrorAction());
         server.setEnablePendingChangesets(configuration.isEnablePendingChangesets());
 
         return server;
     }
 
     public HgVersion getHgVersion(Repository repo) {
         if (this.hgVersion == null) {
             this.hgVersion = VersionCommand.on(repo).execute();
         }
         return this.hgVersion;
     }
 
     @VisibleForTesting
     public List<Server> getServers() {
         return servers;
     }
 
     public int getNumIdleServers() {
         return freeServers.size(); 
     }
 
     // inner types
 
     private final class ServerSupervisor implements Runnable {
         private final Server server;
 
         private ServerSupervisor(Server server) {
             this.server = server;
         }
 
         public void run() {
             if ((System.currentTimeMillis() - server
                     .getLastActiveTime()) > configuration
                     .getServerIdleTime() * 1000) {
                 if (freeServers.remove(server)) {
                     new Thread(new Runnable() {
                         public void run() {
                             stop(server);
                         }
                     }).start();
                 }
                 // Else the server is running a long command and isn't
                 // actually idle.
             }
         }
     }
 }