/*
      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.util.Utils.COLON;
  import static fulmine.util.Utils.COMMA_SPACE;
  import static fulmine.util.Utils.string;
  
  import java.io.IOException;
  import java.net.InetSocketAddress;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.nio.channels.SocketChannel;
  import java.util.List;
  
  import fulmine.context.IFrameworkContext;
  import fulmine.distribution.connection.IConnection;
  import fulmine.distribution.connection.IConnectionBroker;
  import fulmine.distribution.events.ConnectionDestroyedEvent;
  import fulmine.distribution.events.MessageConsumedEvent;
  import fulmine.event.EventSource;
  import fulmine.event.IEvent;
  import fulmine.event.IEventSource;
  import fulmine.event.listener.AbstractEventHandler;
  import fulmine.event.listener.IEventListener;
  import fulmine.event.listener.MultiEventListener;
  import fulmine.util.log.AsyncLog;
  
  /**
   * A {@link SocketChannel} connection to a remote {@link IFrameworkContext}
   * instance. This uses the Java NIO framework.
   * 
   * @author Ramon Servadei
   */
  public abstract class AbstractSocketChannelConnection extends
      TcpConnectionParameters implements IConnection
  {
      /**
       * Handles the {@link MessageConsumedEvent}s and decrements the count of
       * messages that are awaiting processing. If
       * {@link TcpConnection#destroyWhenMessagesProcessed} is set and the count
       * of {@link TcpConnection#messagesToProcess} is zero, the handler will
       * finish the destroy sequence of the connection.
       * 
       * @author Ramon Servadei
       */
      private final class MessageConsumedEventHandler extends
          AbstractEventHandler<MessageConsumedEvent>
      {
          @Override
          public void handle(MessageConsumedEvent event)
          {
              boolean destroy = false;
              synchronized (AbstractSocketChannelConnection.this)
              {
                  AbstractSocketChannelConnection.this.messagesToProcess--;
                  if (AbstractSocketChannelConnection.this.messagesToProcess == 0
                      && AbstractSocketChannelConnection.this.destroyWhenMessagesProcessed)
                  {
                      destroy = true;
                  }
              }
              if (destroy)
              {
                  // finish the destroy sequence
                  AbstractSocketChannelConnection.this.doComponentDestroy();
              }
          }
  
          @Override
          protected AsyncLog getLog()
          {
              return AbstractSocketChannelConnection.this.getLog();
          }
  
          @Override
          public String toString()
          {
              return getClass().getSimpleName();
          }
      }
  
      /** The socket channel to the remote {@link IConnectionBroker} */
      private SocketChannel socketChannel;
  
      /** The delegate for the {@link IEventSource} operations of this */
     protected IEventSource eventSourceDelegate;
 
     /**
      * Flag to indicate the socket direction, by default connections are
      * outbound
      */
     private boolean outbound = true;
 
     /**
      * Set when the connection is established; this happens when both contexts
      * know each others identity. The context making the outbound connection
      * waits for the connection 'ack' from the other context. The context
      * receiving the connection sets this when it receives the 'con' message
      * from the initiating context.
      */
     private boolean connectionEstablished;
 
     /** The handler for {@link MessageConsumedEvent}s */
     private final IEventListener messageConsumedEventHandler;
 
     /**
      * A flag to inform the {@link MessageConsumedEventHandler} that the
      * connection should be destroyed when all messages have been consumed.
      * 
      * @see #messagesToProcess
      */
     protected volatile boolean destroyWhenMessagesProcessed;
 
     /** The context to use for raising events */
     private final IFrameworkContext context;
 
     /**
      * Tracks the number of messages received and not yet processed. When the
      * message is processed, a {@link MessageConsumedEvent} is raised that is
      * intercepted by the {@link MessageConsumedEventHandler} which decrements
      * this.
      * <p>
      * Access must be synchronized on the
      * {@link AbstractSocketChannelConnection} instance.
      */
     protected int messagesToProcess;
 
     /**
      * Standard constructor for the parameters of the connection.
      * 
      * @param context
      *            the context this connection services
      * @param identity
      *            the connection identity
      * @param address
      *            the IP address or resolvable host name of the remote context
      * @param port
      *            the port for the socket connection to the remote context
      * @param connectionHashCode
      *            the hashcode of the connection to the remote context
      * @throws IllegalStateException
      *             if the socket could not be constructed
      */
     @SuppressWarnings("unchecked")
     public AbstractSocketChannelConnection(IFrameworkContext context,
         String identity, String address, int port, int connectionHashCode)
     {
         super(identity, connectionHashCode, address, port);
         this.context = context;
         this.messageConsumedEventHandler =
             new MultiEventListener(
                 MessageConsumedEventHandler.class.getSimpleName(),
                 context,
                 AbstractEventHandler.getEventHandlerMappings(this.new MessageConsumedEventHandler()));
     }
 
     // no messages to process...
     protected void doComponentDestroy()
     {
         // log the final part of the destroy sequence
         if (getLog().isInfoEnabled())
         {
             getLog().info("Destroying[2] " + this + " (stacktrace follows)",
                 new Exception());
         }
         if (this.eventSourceDelegate != null)
         {
             this.eventSourceDelegate.destroy();
         }
     }
 
     protected IFrameworkContext getContext()
     {
         return this.context;
     }
 
     @Override
     protected void doStart()
     {
         try
         {
             if (isOutbound())
             {
                 final SocketChannel localSocketChannel = SocketChannel.open();
                 localSocketChannel.configureBlocking(false);
                 // as the channel is in non-blocking mode, we need some extra
                 // work to connect
                 if (!localSocketChannel.connect(new InetSocketAddress(
                     getRemoteHostAddress(), getRemoteHostTcpPort())))
                 {
                     // the connection has not completed, so wait on the selector
                     // until we can complete the connection
                     final Selector selector = Selector.open();
                     final SelectionKey key =
                         localSocketChannel.register(selector,
                             SelectionKey.OP_CONNECT);
                     selector.select();
                     localSocketChannel.finishConnect();
                     key.cancel();
                     selector.close();
                 }
                 if (getLog().isDebugEnabled())
                 {
                     getLog().debug("(->) Connected to " + this);
                 }
                 setSocketChannel(localSocketChannel);
             }
             else
             {
                 if (getLog().isDebugEnabled())
                 {
                     getLog().debug("(<-) Connected to " + this);
                 }
             }
         }
         catch (Exception e)
         {
             this.socketChannel = null;
             final String warning =
                 "Could not create socket to " + getAddress() + COLON
                     + getRemoteHostTcpPort() + ". Calling destroy().";
             destroy();
             throw new RuntimeException(warning, e);
         }
     }
 
     public boolean isConnected()
     {
         return this.socketChannel != null
             && this.socketChannel.socket() != null
             && this.socketChannel.socket().isConnected();
     }
 
     /**
      * Set the direction of the socket connection.
      * 
      * @param outbound
      *            <code>true</code> for an outbound connecting socket (connects
      *            to a server socket of a remote context), <code>false</code>
      *            for an inbound connecting one
      */
     void setOutbound(boolean outbound)
     {
         this.outbound = outbound;
     }
 
     /**
      * Indicates the connecting direction.
      * 
      * @return <code>true</code> if this socket connects to a server socket,
      *         <code>false</code> if it was an inbound connection from another
      *         context
      */
     public boolean isOutbound()
     {
         return outbound;
     }
 
     /**
      * Destroy the internal TCP I/O components.
      */
     @Override
     protected void doDestroy()
     {
         if (getLog().isInfoEnabled())
         {
             getLog().info("Destroying[1] " + this + " (stacktrace follows)",
                 new Exception());
         }
         if (this.socketChannel != null)
         {
             try
             {
                 if (this.socketChannel.socket() != null)
                 {
                     this.socketChannel.socket().getOutputStream().flush();
                     this.socketChannel.socket().shutdownInput();
                     this.socketChannel.socket().shutdownOutput();
                     this.socketChannel.socket().close();
                 }
                 this.socketChannel.close();
             }
             catch (IOException e)
             {
                 if (getLog().isDebugEnabled())
                 {
                     getLog().debug("Could not close " + this, e);
                 }
             }
         }
         synchronized (this)
         {
             if (this.messagesToProcess == 0)
             {
                 doComponentDestroy();
             }
         }
         if (getContext() != null && getContext().isActive())
         {
             getContext().queueEvent(
                 new ConnectionDestroyedEvent(getContext(),
                     this.getRemoteContextIdentity()));
         }
         if (getLog().isInfoEnabled())
         {
             getLog().info("Destroyed " + this);
         }
     }
 
     public void addEvent(IEvent event)
     {
         this.eventSourceDelegate.addEvent(event);
     }
 
     public boolean addListener(IEventListener listener)
     {
         return this.eventSourceDelegate.addListener(listener);
     }
 
     public byte getEventSourceGroupId()
     {
         return this.eventSourceDelegate.getEventSourceGroupId();
     }
 
     public List<IEventListener> getListeners()
     {
         return this.eventSourceDelegate.getListeners();
     }
 
     public boolean removeListener(IEventListener listener)
     {
         return this.eventSourceDelegate.removeListener(listener);
     }
 
     public List<IEventListener> removeListeners()
     {
         return this.eventSourceDelegate.removeListeners();
     }
 
     public String toDetailedString()
     {
         return toString();
     }
 
     public String toIdentityString()
     {
         return super.toString();
     }
 
     protected SocketChannel getSocketChannel()
     {
         return this.socketChannel;
     }
 
     /**
      * Determine if the application level connection handshake has been
      * established. When a transport connection is made, the context making the
      * connection sends out a 'con' message which the other context responds to
      * with an 'ack'. This sequence completes the application level handshake
      * for the connection and exchanges details about each context.
      * <p>
      * Each context marks its end of the connection as established as per the
      * following;
      * <ul>
      * <li>when the context making the outbound connection receives the 'ack'
      * <li>when the context receiving the inbound connection receives the 'con'
      * </ul>
      * 
      * @return <code>true</code> if the connection handshake has been completed
      *         and the details about the remote context are known.
      */
     protected boolean isConnectionHandshakeComplete()
     {
         return this.connectionEstablished;
     }
 
     /**
      * This method is called when all relevant details of the connection have
      * been established.
      * 
      * @see #isConnectionHandshakeComplete()
      */
     protected void connectionHandshakeComplete()
     {
         this.connectionEstablished = true;
         this.eventSourceDelegate = new EventSource(toString());
         this.eventSourceDelegate.start();
         addListener(messageConsumedEventHandler);
     }
 
     void setSocketChannel(SocketChannel socketChannel)
     {
         this.socketChannel = socketChannel;
     }
 
     @Override
     public String toString()
     {
         if (this.socketChannel != null)
         {
             return string(this, getIdentity() + COMMA_SPACE
                 + this.socketChannel.socket() + ", remoteContextHashcode="
                 + getRemoteContextHashCode());
         }
         return super.toString();
     }
 }