/*
    Copyright 2007 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 fulmine.distribution.connection.tcp;
  
  import static fulmine.distribution.connection.tcp.ProtocolMessageConstants.CONNECTION_MSG_TYPE;
  import static fulmine.distribution.connection.tcp.ProtocolMessageConstants.DELIMITER;
  import static fulmine.util.Utils.COLON;
  import static fulmine.util.Utils.EMPTY_STRING;
  import static fulmine.util.Utils.logException;
  
  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 fulmine.AbstractLifeCycle;
  import fulmine.context.IFrameworkContext;
  import fulmine.distribution.connection.IConnectionBroker;
  import fulmine.distribution.connection.IConnectionParameters;
  import fulmine.event.EventFrameExecution;
  import fulmine.model.container.IContainer;
  import fulmine.model.field.IntegerField;
  import fulmine.model.field.StringField;
  import fulmine.protocol.specification.ByteWriter;
  import fulmine.util.io.AbstractSelectionKeyTask;
  import fulmine.util.io.ISelectorTasksStateListener;
  import fulmine.util.io.SelectorTasks;
  import fulmine.util.log.AsyncLog;
  
  /**
   * A connection broker implementation based on TCP/IP sockets. This has a single
   * thread that handles input and output. This should be adequate as a context is
   * usually the source or sink of data.
   * <p>
   * The TCP parameters are controlled by:
   * <ul>
   * <li>{@link #CONTEXT_IP_ADDRESS}
   * <li>{@link #CONTEXT_TCP_PORT}
   * <li>{@link #CONTEXT_RCV_BUFFER}
   * </ul>
   * 
   * @author Ramon Servadei
   */
  public final class TcpConnectionBroker extends AbstractLifeCycle implements
      IConnectionBroker
  {
      final static AsyncLog LOG = new AsyncLog(TcpConnectionBroker.class);
  
      /**
       * The system property for the IP address of the context server socket. On a
       * multi-homed host, this is useful to control which address the server
       * socket binds to. Use <code>null</code> to accept on all addresses the
       * host uses.
       */
      public static final String CONTEXT_IP_ADDRESS = "context.ip.address";
  
      /** The system property for the TCP port */
      public static final String CONTEXT_TCP_PORT = "context.tcp.port";
  
      /** The default for the TCP port */
      public static final int DEFAULT_CONTEXT_TCP_PORT = 16207;
  
      /**
       * 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 = "context.tcp.buffer.rcv";
  
      /** The host name */
      public final static String HOST_NAME;
      static
      {
          String name = null;
          try
          {
              name = InetAddress.getLocalHost().getHostName();
          }
          catch (UnknownHostException e)
          {
              LOG.fatal("Could not get host name", e);
              System.exit(1);
          }
          HOST_NAME = name;
     }
 
     /** The IP address for the TCP/IP socket of this context */
     private final String address;
 
     /** The port number for the TCP/IP socket of this context */
     private final int port;
 
     /** The server socket channel of this context */
     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;
 
     /** The context to use when subscribing for events */
     final IFrameworkContext context;
 
     /**
      * Construct the {@link TcpConnectionBroker} using parameters for the
      * identity, IP address and TCP port.
      * 
      * @param context
      *            The context to use when subscribing for events
      * @param address
      *            the IP address of the host of this {@link TcpConnectionBroker}
      *            , <code>null</code> for the host address of the local host
      * @param port
      *            the TCP port for this{@link TcpConnectionBroker},
      *            <code>0</code> for {@link #DEFAULT_CONTEXT_TCP_PORT}
      */
     public TcpConnectionBroker(IFrameworkContext context, String address,
         int port)
     {
         super();
         this.context = context;
         this.address =
             address == null ? TcpConnectionBroker.HOST_NAME : address;
         this.port =
             port <= 0 ? TcpConnectionBroker.DEFAULT_CONTEXT_TCP_PORT : port;
         this.selectorTasks = new SelectorTasks();
         this.connectionAcceptTask = new ConnectionAcceptTask();
         this.selectorTasks.setStateListener(this.connectionAcceptTask);
 
         if (getLog().isInfoEnabled())
         {
             getLog().info("Created " + this);
         }
     }
 
     /**
      * Default constructor that reads system properties for the IP address and
      * TCP port to use.
      * 
      * @param context
      *            The context to use when subscribing for events
      * @see #CONTEXT_IP_ADDRESS
      * @see #CONTEXT_TCP_PORT
      * @see #DEFAULT_CONTEXT_TCP_PORT
      */
     public TcpConnectionBroker(IFrameworkContext context)
     {
         this(context, System.getProperty(
             TcpConnectionBroker.CONTEXT_IP_ADDRESS,
             TcpConnectionBroker.HOST_NAME),
             Integer.parseInt(System.getProperty(
                 TcpConnectionBroker.CONTEXT_TCP_PORT, EMPTY_STRING
                     + TcpConnectionBroker.DEFAULT_CONTEXT_TCP_PORT)));
     }
 
     public String getAddress()
     {
         return this.address;
     }
 
     public int getPort()
     {
         return this.port;
     }
 
     @Override
     protected void doDestroy()
     {
         if (getLog().isInfoEnabled())
         {
             getLog().info("Destroying " + this);
         }
         if (this.serverSocketChannel != null)
         {
             try
             {
                 this.serverSocketChannel.close();
             }
             catch (IOException e)
             {
                 if (getLog().isDebugEnabled())
                 {
                     getLog().debug("Could not close " + this, e);
                 }
             }
         }
         this.selectorTasks.destroy();
         if (getLog().isInfoEnabled())
         {
             getLog().info("Destroyed " + this);
         }
     }
 
     // synchronized to ensure only 1 thread creates a connection at a time
     public synchronized void connect(IConnectionParameters connectionParameters)
     {
         TcpConnection connection = null;
 
         if (!isActive())
         {
             return;
         }
 
         TcpConnectionParameters parameters =
             (TcpConnectionParameters) connectionParameters;
 
         connection =
             new TcpConnection(this.context, this, parameters.getIdentity(),
                 parameters.getRemoteHostAddress(),
                 parameters.getRemoteHostTcpPort(),
                 parameters.getRemoteContextHashCode());
         if (getLog().isInfoEnabled())
         {
             getLog().info("(->) Connecting to outbound " + connection);
         }
         connection.start();
 
         if (connection.getSocketChannel() != null)
         {
             registerConnection(connection, connection.getSocketChannel());
             if (getLog().isInfoEnabled())
             {
                 getLog().info("(->) Registered outbound " + connection);
             }
             // send the connection 'handshake'
             connection.send(ByteWriter.getBytes(CONNECTION_MSG_TYPE + DELIMITER
                 + this.context.getIdentity() + DELIMITER
                 + this.context.getContextHashCode()));
         }
         else
         {
             if (getLog().isInfoEnabled())
             {
                 getLog().info("No socket for " + connection);
             }
         }
     }
 
     private void registerConnection(TcpConnection tcpConnection,
         SocketChannel socketChannel)
     {
         this.selectorTasks.register(SelectionKey.OP_READ, socketChannel,
             tcpConnection);
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     @Override
     protected void doStart()
     {
         try
         {
             if (getLog().isInfoEnabled())
             {
                 getLog().info("Starting " + this);
             }
             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.openSelector();
             this.connectionThread =
                 new Thread(this.selectorTasks.getRunnable(), this.toString());
             this.connectionThread.setDaemon(true);
             this.connectionThread.start();
         }
         catch (IOException e)
         {
             getLog().fatal("Could not open channel for " + this, e);
             // super fatal
             System.exit(1);
         }
         if (getLog().isInfoEnabled())
         {
             getLog().info("Started " + this);
         }
         final IContainer systemInfo = this.context.getSystemInfo();
         systemInfo.beginFrame(new EventFrameExecution());
         try
         {
             systemInfo.add(new StringField(CONTEXT_IP_ADDRESS, getAddress()));
             systemInfo.add(new IntegerField(CONTEXT_TCP_PORT, getPort()));
         }
         finally
         {
             systemInfo.endFrame();
         }
     }
 
     SelectorTasks getSelectorTasks()
     {
         return this.selectorTasks;
     }
 
     /**
      * Task that handles new socket connections to this context's server socket.
      * It creates a {@link TcpConnection} and registers this against the
      * {@link SocketChannel} that wraps the underlying new socket connection.
      * 
      * @author Ramon Servadei
      * 
      */
     private final class ConnectionAcceptTask extends AbstractSelectionKeyTask
         implements ISelectorTasksStateListener
     {
         public ConnectionAcceptTask()
         {
             super();
         }
 
         public void run()
         {
             if (TcpConnectionBroker.this.isActive()
                 && getSelectionKey().isAcceptable())
             {
                 try
                 {
                     SocketChannel conn =
                         TcpConnectionBroker.this.serverSocketChannel.accept();
 
                     if (LOG.isInfoEnabled())
                     {
                         LOG.info("(<-) Accepted inbound " + conn);
                     }
                     conn.configureBlocking(false);
                     final String address =
                         conn.socket().getInetAddress().getHostName();
                     final int port = conn.socket().getPort();
                     final TcpConnection connection =
                         new TcpConnection(TcpConnectionBroker.this.context,
                             TcpConnectionBroker.this, "(pending...)", address,
                             port, 0);
                     connection.setOutbound(false);
                     connection.setSocketChannel(conn);
                     connection.start();
                     registerConnection(connection,
                         connection.getSocketChannel());
                     if (LOG.isInfoEnabled())
                     {
                         LOG.info("(<-) Registered inbound " + conn);
                     }
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), this
                         + " could not accept connection", e);
                 }
             }
         }
 
         public void selectorOpened(SelectorTasks tasks)
         {
             if (TcpConnectionBroker.this.isActive())
             {
                 // (re)register
                 TcpConnectionBroker.this.selectorTasks.register(
                     SelectionKey.OP_ACCEPT,
                     TcpConnectionBroker.this.serverSocketChannel,
                     TcpConnectionBroker.this.connectionAcceptTask);
             }
         }
     }
 
     @Override
     public String toString()
     {
         return getClass().getSimpleName() + "<" + this.context.getIdentity()
             + COLON + this.address + COLON + this.port + ">";
     }
 }