package org.sourceforge.jemm.comm.server;
   
   import java.util.HashMap;
   import java.util.HashSet;
   import java.util.Map;
   import java.util.Set;
   import java.util.concurrent.CountDownLatch;
   import java.util.concurrent.ExecutorService;
   import java.util.concurrent.Executors;
  
  import org.apache.log4j.Logger;
  import org.sourceforge.jemm.comm.connection.Connection;
  import org.sourceforge.jemm.comm.connection.ConnectionException;
  import org.sourceforge.jemm.comm.connection.ServerConnectionFactory;
  import org.sourceforge.jemm.comm.shared.IFUtilities;
  import org.sourceforge.jemm.comm.shared.RPCHandler;
  
  /**
   * The RPCServer offers RPC interfaces to connecting clients.  The server can handle multiple
   * client connections.
   *
   * See {@link org.sourceforge.jemm.comm.example.echo.EchoClient}/{@link 
   * org.sourceforge.jemm.comm.example.echo.EchoServer} for an example of usage.
   * 
   * <i><P>Released under the <a href="http://www.apache.org/licenses/LICENSE-2.0.html">Apache License V2.0 license</a> 
   * by the <a href="http://jemm.sourceforge.net">JEMM Project</a></i>
   *
   * @author Rory Graves
   */
  public class RPCServer extends Thread {
      protected static final Logger LOG = Logger.getLogger(RPCServer.class);
  
      /** The map of clientIds to active clients */
      protected Map<RPCClientId, ServerThread> clientThreads = new HashMap<RPCClientId, ServerThread>();
  
      /** A latch to ensure server is ready when initialisation call returns. */
      protected CountDownLatch initialisationLatch = new CountDownLatch(1);
  
      /** Used for graceful shutdown. */
      protected boolean closing = false;
  
      /** The thread pool for processing client call requests (shared among all clients) */
      protected ExecutorService msgProccessingThreadPool;
  
      protected final ServerConnectionFactory connectionFactory;
  
      /** A map of the interfaces implemented by the server and their corresponding interface implementations. */
      HashMap<Class<?>, Object> serverIFMap = new HashMap<Class<?>, Object>();
  
      ClientConnectionListener clientConnListener;
  
      /**
       * Creates an RPCServer instance
       *
       * @param connectionFactory The connection factory for accepting client connections.
       * @param executorService The executor used to service requests.
       */
      public RPCServer(ServerConnectionFactory connectionFactory,ExecutorService executorService) {
      	this.connectionFactory = connectionFactory;
          msgProccessingThreadPool = executorService;
      }
  
      /**
       * Creates an RPCServer instance
       *
       * @param connectionFactory The connection factory for accepting client connections.
       * @param noProcessingThreads The number of processing threads.
       */
      public RPCServer(ServerConnectionFactory connectionFactory,int noProcessingThreads) {
      	this(connectionFactory,Executors.newFixedThreadPool(noProcessingThreads));
      }
  
      /**
       * Notification that a given client has disconnected. The client is removed from the active
       * pool.
       *
       * @param clientId The assigned id of the disconnected client.
       */
      public void clientDisconnected(final RPCClientId clientId) {
          LOG.info("client " + clientId + " disconnected");
  
          final ClientConnectionListener listener = this.clientConnListener;
          if(listener != null)
              listener.clientDisconnected(clientId);
  
          synchronized (clientThreads) {
              clientThreads.remove(clientId);
          }
      }
  
      /**
       * Used by called methods to find out ID of calling client.
       * This uses a ThreadLocal which is set before the server method is invoked.
       * @return The id of the caller.
       */
      public RPCClientId getClientId() {
          return (RPCClientId) RPCHandler.getConnectionId();
      }
  
     /**
      * Retrieve a proxy for the given interface on the given client.
      * @param clientId The id of the client.
      * @param ifClass The required interface class.
      * @return An object proxying the interface to the given client.
      */
     public Object getClientIF(RPCClientId clientId,
             Class<?> ifClass) {
         ServerThread st = clientThreads.get(clientId);
         return st.getClientIF(ifClass);
     }
 
     /**
      * @param clientId The id of the new client.
      * @param clientHostname The clients hostname.
      */
     public void notifyNewClient(RPCClientId clientId,String clientAddress) {
         ClientConnectionListener listener = this.clientConnListener;
         if(listener != null)
             listener.clientConnected(clientId,clientAddress);
         else
             LOG.debug("No listener set for new client notifications");
     }
 
     /**
      * Register an interface for RPC calls by clients.
      * @param targetIF The interface to offer to clients.
      * @param targetIFImpl The server side implementation of the given interface.
      */
     public void registerInterface(Class<?> targetIF,
             Object targetIFImpl) {
         IFUtilities.validateInterface(targetIF);
         
         if(!targetIF.isInstance(targetIFImpl))
             throw new IllegalArgumentException("Target impl not instance of interface");
 
         serverIFMap.put(targetIF, targetIFImpl);
     }
 
     @Override
     public void run() {
         try {
             LOG.info("Server started");
             connectionFactory.initialise();
             LOG.info("Listening for client connections");
             initialisationLatch.countDown();
             
             while (true) {
                 Connection connection = connectionFactory.getClientConnection();
                 RPCClientId clientId = new RPCClientId();
                 LOG.info("connection from " + connection.getConnectionName() + " assigned id=" + clientId);
                 ServerThread st = new ServerThread(this, connection, clientId,serverIFMap,msgProccessingThreadPool);
                 synchronized (clientThreads) {
                     clientThreads.put(clientId, st);
                 }
                 
                 // starts reading messages after client thread has been added to list to avoid race
                 // condition.
                 st.start();
             }
         } catch (ConnectionException e) {
             if (!closing)
                 LOG.error("Unexpected exception whilst accepting connections",e);
         }
 
         LOG.info("Server terminated");
     }
 
     /**
      * Sets the client listener to the given listener.
      * @param listener The listener to receive disconnection events.
      */
     public void setClientListener(ClientConnectionListener listener) {
         this.clientConnListener = listener;
     }
 
     /**
      * Request by the system for the server to be shutdown.
      */
     public void shutdown() {
         closing = true;
         try {
             LOG.info("Shutting down");
 
             // close the server socket to stop accepting new connections
             connectionFactory.close();
             Set<ServerThread> toShutdown = new HashSet<ServerThread>();
             toShutdown.addAll(clientThreads.values());
 
             // notify any remaining clients to shutdown
             LOG.info("Notifying client threads");
             for (ServerThread clientThread : toShutdown)
                 clientThread.shutdown();
 
             for (ServerThread clientThread : toShutdown)
                 clientThread.waitForShutdown();
             
             LOG.info("Client threads complete");
             msgProccessingThreadPool.shutdown();
 
         } catch (ConnectionException ce) {
             LOG.warn("Exception thrown whilst terminating server", ce);
         }
         LOG.info("Shutdown complete");
 
     }
 
     /**
      * This method will pause the caller until the server is ready to start
      * accepting connections.  If the server is already running it will return
      * immediately.  Mostly used for testing when a client and server are
      * run in the same VM to avoid the client connecting before the server is
      * ready.
      */
     public void waitForInitialisation() {
         try {
             initialisationLatch.await();
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     }
 }