/*
      Copyright 2010 Ramon Servadei
   
      Licensed under the Apache License, Version 2.0 (the "License");
      you may not use this file except 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.
   */
  package mirrormap.nio;
  
  import java.io.IOException;
  import java.net.InetAddress;
  import java.net.InetSocketAddress;
  import java.net.UnknownHostException;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.ServerSocketChannel;
  import java.nio.channels.SocketChannel;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ThreadFactory;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import mirrormap.Utils;
  import mirrormap.lifecycle.AbstractLifeCycle;
  import mirrormap.nio.IConnectionEndPointReceiver.IFactory;
  
  /**
   * A server that opens a TCP server socket and spawns
   * {@link TcpConnectionEndPoint} instances to handle each of the client
   * connections.
   * 
   * @author Ramon Servadei
   */
  public class TcpServer extends AbstractLifeCycle
  {
      private final static Logger LOGGER =
          Logger.getLogger(TcpServer.class.getName());
  
      /**
       * The system property for overriding the default receive buffer size for
       * the server port. This must be specified in units of bytes.
       */
      public static final String CONTEXT_RCV_BUFFER = "TcpServer.bufferSize";
  
      /** The default TCP port */
      public static final int DEFAULT_TCP_PORT = 16025;
  
      /** The host name */
      public final static String HOST_NAME;
  
      static
      {
          String name = null;
          try
          {
              name = InetAddress.getLocalHost().getHostName();
          }
          catch (UnknownHostException e)
          {
              LOGGER.log(Level.SEVERE, "Could not get host name", e);
              System.exit(1);
          }
          HOST_NAME = name;
      }
  
      /**
       * Task that handles new connections to this server. It creates a
       * {@link TcpConnectionEndPoint} for each connection.
       * 
       * @author Ramon Servadei
       */
      private final class ConnectionAcceptTask implements
          ISelectorTasksStateListener, ISelectionKeyTask
      {
  
          /** The selection key */
          private SelectionKey selectionKey;
  
          public ConnectionAcceptTask()
          {
              super();
          }
  
          public void run()
          {
              if (TcpServer.this.isActive() && getSelectionKey().isAcceptable())
              {
                  try
                  {
                      SocketChannel socketChannel =
                          TcpServer.this.serverSocketChannel.accept();
                     if (LOGGER.isLoggable(Level.INFO))
                     {
                         LOGGER.info("(<-) Accepted inbound " + socketChannel);
                     }
                     socketChannel.configureBlocking(false);
                     final String address =
                         socketChannel.socket().getInetAddress().getHostName();
                     final int port = socketChannel.socket().getPort();
                     final IConnectionEndPointReceiver createEndPointRecevier =
                         TcpServer.this.endPointReceiverFactory.createEndPointRecevier();
                     final TcpConnectionEndPoint endPoint =
                         new TcpConnectionEndPoint(createEndPointRecevier,
                             TcpServer.this.selectorTasks,
                             TcpServer.this.writer, address, port);
                     createEndPointRecevier.setEndPointConnection(endPoint);
                     endPoint.setOutbound(false);
                     endPoint.setSocketChannel(socketChannel);
                     endPoint.start();
                 }
                 catch (Exception e)
                 {
                     Utils.logException(LOGGER, Utils.safeToString(this)
                         + " could not accept connection", e);
                 }
             }
         }
 
         public void selectorOpened(SelectorTasks tasks)
         {
             if (TcpServer.this.isActive()
                 && TcpServer.this.serverSocketChannel != null
                 && TcpServer.this.connectionAcceptTask != null)
             {
                 // (re)register
                 TcpServer.this.selectorTasks.register(SelectionKey.OP_ACCEPT,
                     TcpServer.this.serverSocketChannel,
                     TcpServer.this.connectionAcceptTask);
             }
         }
 
         public SelectionKey getSelectionKey()
         {
             return this.selectionKey;
         }
 
         public void setSelectionKey(SelectionKey selectionKey)
         {
             this.selectionKey = selectionKey;
         }
 
         public final void destroy()
         {
             this.selectionKey.cancel();
         }
 
         @Override
         public String toString()
         {
             return getClass().getSimpleName();
         }
     }
 
     /** The IP address for the TCP/IP socket of this server */
     private final String address;
 
     /** The port number for the TCP/IP socket of this server */
     private final int port;
 
     /** The server socket channel of this server */
     ServerSocketChannel serverSocketChannel;
 
     /** The selector tasks */
     private final SelectorTasks selectorTasks;
 
     /** The task that handles new socket connections to the server socket */
     private final ConnectionAcceptTask connectionAcceptTask;
 
     /** The thread that handles all the socket IO activity */
     private Thread connectionThread;
 
     /** Handles socket writing activity */
     private final ExecutorService writer =
         Executors.newFixedThreadPool(1, new ThreadFactory()
         {
             public Thread newThread(Runnable r)
             {
                 Thread thread =
                     new Thread(r, TcpServer.class.getSimpleName()
                         + " NIO writer");
                 thread.setDaemon(true);
                 return thread;
             }
         });
 
     /** Creates {@link IConnectionEndPointReceiver} instances per client socket */
     private final IFactory endPointReceiverFactory;
 
     /**
      * Construct the server with the factory that will create the
      * {@link IConnectionEndPointReceiver} per client socket connection and the
      * server socket address and port
      * 
      * @param endPointReceiverFactory
      *            the factory to create the {@link IConnectionEndPointReceiver}
      *            for each client socket
      * @param address
      *            the server socket address or host name
      * @param port
      *            the server socket TCP port
      */
     public TcpServer(
         IConnectionEndPointReceiver.IFactory endPointReceiverFactory,
         String address, int port)
     {
         super();
         this.address = address == null ? TcpServer.HOST_NAME : address;
         this.port = port <= 0 ? TcpServer.DEFAULT_TCP_PORT : port;
         this.selectorTasks = new SelectorTasks();
         this.connectionAcceptTask = new ConnectionAcceptTask();
         this.selectorTasks.setStateListener(this.connectionAcceptTask);
         this.endPointReceiverFactory = endPointReceiverFactory;
     }
 
     @Override
     public String toString()
     {
         return Utils.string(this, this.address + ":" + this.port);
     }
 
     @Override
     protected void doDestroy()
     {
         this.writer.shutdownNow();
         this.selectorTasks.destroy();
         try
         {
             this.serverSocketChannel.close();
         }
         catch (IOException e)
         {
             Utils.logException(LOGGER, "Could not close "
                 + Utils.safeToString(this.serverSocketChannel), e);
         }
     }
 
     @Override
     protected void doStart()
     {
         try
         {
             this.serverSocketChannel = ServerSocketChannel.open();
             this.serverSocketChannel.configureBlocking(false);
             if (System.getProperty(CONTEXT_RCV_BUFFER) != null)
             {
                 this.serverSocketChannel.socket().setReceiveBufferSize(
                     Integer.parseInt(System.getProperty(CONTEXT_RCV_BUFFER)));
             }
             final InetSocketAddress endpoint =
                 new InetSocketAddress(this.address, this.port);
             this.serverSocketChannel.socket().bind(endpoint);
 
             this.selectorTasks.start();
             this.connectionThread =
                 new Thread(this.selectorTasks.getRunnable(), this.toString());
             this.connectionThread.setDaemon(true);
             this.connectionThread.start();
         }
         catch (IOException e)
         {
             LOGGER.log(Level.SEVERE, "Could not open channel for " + this, e);
             // super fatal
             System.exit(1);
         }
     }
 }