/*
    * Created on 17 Nov 2007
    * 
    */
   package org.sourceforge.jemmrpc.server;
   
   import java.io.IOException;
   import java.net.ServerSocket;
   import java.net.Socket;
  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.jemmrpc.example.echo.EchoClient;
  import org.sourceforge.jemmrpc.example.echo.EchoServer;
  import org.sourceforge.jemmrpc.shared.IFUtilities;
  import org.sourceforge.jemmrpc.shared.Message;
  import org.sourceforge.jemmrpc.shared.RPCHandler;
  
  /**
   * The RPCServer offers RPC interfaces to connecting clients. The server can handle multiple client
   * connections.
   * 
   * See {@link EchoClient}/{@link EchoServer} for an example of usage.
   * 
   * @author Rory Graves
   * 
   */
  public class RPCServer extends Thread
  {
      protected static final Logger logger = Logger.getLogger(RPCServer.class);
  
      /** The map of clientIds to active clients */
      protected Map<ClientId, ServerThread> clientThreads = new HashMap<ClientId, 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;
  
      /** The port the server listens on. */
      protected int port;
  
      /** The server socket which listens for client connections. */
      protected ServerSocket serverSocket;
  
      /**
       * A map of the interfaces implemented by the server and their corresponding interface
       * implementations.
       */
      HashMap<Class<?>, Object> serverIFMap = new HashMap<Class<?>, Object>();
  
      ClientConnectionListener clientConnListener;
  
      /**
       * 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 ClientId getClientId()
      {
          return (ClientId) 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(ClientId clientId, Class<?> ifClass)
      {
          final ServerThread st = clientThreads.get(clientId);
          return st.getClientIF(ifClass);
      }
  
      /**
       * Creates an RPCServer instance
       * 
       * @param port The socket to listen for connections on
       * @param noProcessingThreads The number of processing threads.
       */
      public RPCServer(int port, int noProcessingThreads)
      {
          this.port = port;
          msgProccessingThreadPool = Executors.newFixedThreadPool(noProcessingThreads);
      }
  
     /**
      * Creates an RPCServer instance
      * 
      * @param port The socket to listen for connections on
      * @param executorService The executor used to service requests.
      */
     public RPCServer(int port, ExecutorService executorService)
     {
         this.port = port;
         msgProccessingThreadPool = executorService;
     }
 
     /**
      * 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 ClientId clientId)
     {
         logger.info("client " + clientId + " disconnected");
 
         final ClientConnectionListener listener = this.clientConnListener;
         if (listener != null)
             listener.clientDisconnected(clientId);
 
         synchronized (clientThreads)
         {
             clientThreads.remove(clientId);
         }
     }
 
     /**
      * Process an incoming message.
      * 
      * @param clientId The ID of the sending client.
      * @param message The received message.
      */
     public void processMessage(final ClientId clientId, final Message message)
     {
     }
 
     /**
      * 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 (final InterruptedException e)
         {
             e.printStackTrace();
         }
     }
 
     @Override
     public void run()
     {
         try
         {
             logger.info("Server started");
             serverSocket = new ServerSocket(port);
             logger.info("Listening on port: " + port);
             initialisationLatch.countDown();
             while (true)
             {
                 final Socket s = serverSocket.accept();
                 final ClientId clientId = new ClientId();
                 logger.info("connection from " + s.getInetAddress() + ":" + s.getPort()
                         + " assigned id=" + clientId);
                 final ServerThread st = new ServerThread(this, s, 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 (final IOException e)
         {
             if (!closing)
                 logger.error("Unexpected exception whilst accepting socket connections",
                         e);
         }
 
         logger.info("Server terminated");
     }
 
     /**
      * Request by the system for the server to be shutdown.
      */
     public void shutdown()
     {
         closing = true;
         try
         {
             logger.info("Shutting down");
 
             // close the server socket to stop accepting new connections
             serverSocket.close();
             msgProccessingThreadPool.shutdown();
             final Set<ServerThread> toShutdown = new HashSet<ServerThread>();
             toShutdown.addAll(clientThreads.values());
 
             // notify any remaining clients to shutdown
             logger.info("Notifying client threads");
             for (final ServerThread clientThread : toShutdown)
                 clientThread.shutdown();
 
             for (final ServerThread clientThread : toShutdown)
                 clientThread.waitForShutdown();
 
             logger.info("Client threads complete");
         } catch (final IOException ioe)
         {
             logger.warn("Exception thrown whilst terminating server", ioe);
         }
         logger.info("Shutdown complete");
 
     }
 
     /**
      * 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);
     }
 
     /**
      * Sets the client listener to the given listener.
      * 
      * @param listener The listener to receive disconnection events.
      */
     public void setClientListener(ClientConnectionListener listener)
     {
         this.clientConnListener = listener;
     }
 
     /**
      * @param clientId The id of the new client.
      * @param clientHostname The clients hostname.
      * @param clientPort The clients port.
      */
     public void notifyNewClient(ClientId clientId, String clientHostname, int clientPort)
     {
         final ClientConnectionListener listener = this.clientConnListener;
         if (listener != null)
             listener.clientConnected(clientId, clientHostname, clientPort);
         else
             logger.debug("No listener set for new client notifications");
     }
 }