package fulmine.distribution.connection.tcp;
   
   import static fulmine.distribution.connection.tcp.ProtocolMessageConstants.CONNECTION_MSG_TYPE;
   import static fulmine.distribution.connection.tcp.ProtocolMessageConstants.CONN_ACK_MSG_TYPE;
   import static fulmine.distribution.connection.tcp.ProtocolMessageConstants.DELIMITER;
   import static fulmine.util.Utils.logException;
   
   import java.io.IOException;
   import java.nio.ByteBuffer;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.SocketChannel;
  import java.util.Arrays;
  import java.util.Iterator;
  import java.util.List;
  
  import fulmine.context.IFrameworkContext;
  import fulmine.distribution.connection.IConnection;
  import fulmine.distribution.events.ConnectionAvailableEvent;
  import fulmine.distribution.events.ConnectionDestroyedEvent;
  import fulmine.distribution.events.MessageConsumedEvent;
  import fulmine.distribution.events.MessageEvent;
  import fulmine.event.IEventSource;
  import fulmine.event.listener.IEventListener;
  import fulmine.protocol.specification.ByteConstants;
  import fulmine.protocol.specification.ByteReader;
  import fulmine.protocol.specification.ByteWriter;
  import fulmine.util.array.ArrayUtils;
  import fulmine.util.collection.CollectionFactory;
  import fulmine.util.io.DelegatingSelectionKeyTask;
  import fulmine.util.io.ISelectionKeyTask;
  import fulmine.util.log.AsyncLog;
  
  /**
   * Handles the socket processing for a single connection to a remote
   * {@link IFrameworkContext}.
   * <p>
   * Received TCP messages from the remote context may be concatenated together;
   * e.g. the remote context sends "conn:lasers" then "syn". These may be received
   * by this connection as a single message "conn:laserssyn". To be able to
   * distinguish these messages, each message is prefixed with the byte length of
   * the application message. This prefix is always 4 bytes, big-endian; e.g. the
   * above example would resolve to "11conn:lasers3syn".
   * <p>
   * This generates the following events:
   * <ul>
   * <li>{@link MessageEvent}
   * <li>{@link ConnectionAvailableEvent}
   * <li>{@link ConnectionDestroyedEvent}
   * </ul>
   * <p>
   * When this object is destroyed via the {@link #destroy()} method, this only
   * closes down the TCP IO resources and signals that only when all pending
   * {@link MessageEvent}s have been processed should this object's
   * {@link IEventSource} functions be terminated; this will then remove any
   * {@link IEventListener}s that would normally receive messages from this
   * connection.
   * 
   * @author Ramon Servadei
   */
  public final class TcpConnection extends AbstractSocketChannelConnection
      implements IEventSource, IConnection, ISelectionKeyTask
  {
      private final static AsyncLog LOG = new AsyncLog(TcpConnection.class);
  
      /**
       * The byte representation of the {@link CONNECTION_MSG_TYPE}
       */
      private final static byte[] CONNECTION_MSG_TYPE_BYTES =
          ByteWriter.getBytes(CONNECTION_MSG_TYPE);
  
      /**
       * The byte representation of the {@link CONN_ACK_MSG_TYPE}
       */
      private final static byte[] CONN_ACK_MSG_TYPE_BYTES =
          ByteWriter.getBytes(CONN_ACK_MSG_TYPE);
  
      /** Delegate {@link ISelectionKeyTask} */
      private DelegatingSelectionKeyTask selectionKeyTask;
  
      /** The pending messages to send */
      final List<ByteBuffer> messages = CollectionFactory.newList();
  
      /** The receive buffer, 65535 bytes */
      private final static byte[] recvBuffer = new byte[65535];
  
      /** The TCP connection broker */
      private final TcpConnectionBroker broker;
  
      /**
       * Standard constructor
       * 
       * @param broker
       *            the {@link TcpConnectionBroker}
       * @param remoteContextIdentity
       *            the identity of the remote context this connection connects to
       * @param address
       *            the TCP address of the server port the context exposes to
       *            other remote contexts
       * @param port
      *            the TCP server port the context exposes to other remote
      *            contexts
      * @param connectionHashCode
      *            the hashcode of the connection to the remote context
      */
     public TcpConnection(IFrameworkContext context, TcpConnectionBroker broker,
         String remoteContextIdentity, String address, int port,
         int connectionHashCode)
     {
         super(context, remoteContextIdentity, address, port, connectionHashCode);
         this.broker = broker;
         this.selectionKeyTask = new DelegatingSelectionKeyTask(this);
     }
 
     /**
      * Add the data to the pending messages list
      * 
      * @param data
      *            the data to add to the pending messages
      */
     public void send(byte[] data)
     {
         synchronized (this)
         {
             byte[] actual = encode(data);
 
             final ByteBuffer wrapped = ByteBuffer.wrap(actual);
             messages.add(wrapped);
             // we're interested in writing now
             getSelectionKey().interestOps(
                 getSelectionKey().interestOps() | SelectionKey.OP_WRITE);
             getSelectionKey().selector().wakeup();
         }
     }
 
     /**
      * Add a 4 byte prefix to the data that includes the length of the data as a
      * big-endian integer.
      * 
      * @param data
      *            the data
      * @return an array equal in length to <code>4 + data.length</code> with a
      *         copy of data starting from element 4
      */
     static byte[] encode(byte[] data)
     {
         // add the integer length of the data, always the first 4 bytes
         byte[] actual = new byte[data.length + 4];
         System.arraycopy(data, 0, actual, 4, data.length);
         int startFrom = 0;
         for (int i = 4; i > 0; i--)
         {
             actual[startFrom++] =
                 (byte) (data.length >>> ByteConstants.bitShiftForByteOrdinal[i]);
         }
         return actual;
     }
 
     /**
      * Split the data into its sub-messages. Messages have a 4 byte prefix that
      * is the message length in bytes, excluding the 4 bytes, as a big-endian
      * integer. This allows multiple messages to be received as a single TCP
      * frame.
      * 
      * @param message
      *            the full TCP frame data
      * @return a list of the individual messages contained in the message
      *         argument
      */
     static List<byte[]> decode(byte[] message)
     {
         // the TCP message could be a concatenation of multiple ones
         // the format is: <4 bytes len><subdata><4 bytes len><subdata>
         List<byte[]> values = CollectionFactory.newList();
         int lenPtr = 0;
         int len = 0;
         while (lenPtr < message.length)
         {
             len = ByteReader.readInteger(message, lenPtr, 4);
             // copy the sub data into the new array
             byte[] subData = new byte[len];
             lenPtr += 4;
             System.arraycopy(message, lenPtr, subData, 0, len);
             lenPtr += len;
             values.add(subData);
         }
         return values;
     }
 
     public void run()
     {
         if (isActive())
         {
             if (getSelectionKey().isWritable())
             {
                 write();
             }
             if (getSelectionKey().isReadable())
             {
                 read();
             }
         }
     }
 
     /**
      * Reads data from the socket
      */
     private void read()
     {
         try
         {
             final ByteBuffer recv = ByteBuffer.wrap(recvBuffer);
             final SocketChannel socketChannel =
                 ((SocketChannel) getSelectionKey().channel());
             final int size = socketChannel.read(recv);
             if (size == -1)
             {
                 /*
                  * end of stream reached (see
                  * ReadableByteChannel.read(ByteBuffer dst)
                  */
                 throw new IOException("end-of-stream");
             }
             final byte[] bufferData = new byte[recv.position()];
             System.arraycopy(recv.array(), 0, bufferData, 0, recv.position());
             // find any sub-messages
             final List<byte[]> decoded = decode(bufferData);
             for (byte[] byteData : decoded)
             {
                 if (getLog().isTraceEnabled())
                 {
                     final String stringData =
                         new String(byteData, ByteConstants.ENCODING).trim();
                     getLog().trace(
                         this + " received '"
                             + Arrays.toString(stringData.getBytes()) + "'");
                 }
                 if (!isConnectionHandshakeComplete())
                 {
                     if (ArrayUtils.startsWith(CONNECTION_MSG_TYPE_BYTES,
                         byteData))
                     {
                         final String data =
                             new String(byteData, ByteConstants.ENCODING).trim();
                         // this is where we find out the identity of the
                         // connecting remote context
                         final String[] fields = data.split(DELIMITER);
                         setIdentity(fields[1]);
                         setRemoteContextHashCode(Integer.parseInt(fields[2]));
                         connectionHandshakeComplete();
                         // send the ack
                         send(ByteWriter.getBytes(CONN_ACK_MSG_TYPE + DELIMITER
                             + getContext().getIdentity() + DELIMITER
                             + getContext().getContextHashCode()));
 
                         // notify with the new connection
                         getContext().queueEvent(
                             new ConnectionAvailableEvent(getContext(), this));
                     }
                     else
                     {
                         if (ArrayUtils.startsWith(CONN_ACK_MSG_TYPE_BYTES,
                             byteData))
                         {
                             /*
                              * retrieve the identity and remote context hashcode
                              * - if we reconnect to a remote context that has
                              * bounced, it will have a new hashcode but this
                              * local context may not have received the new
                              * connection details with the new hashcode. To
                              * prevent the situation where the newly made
                              * connection will be severed because the hashcodes
                              * would be different, we always set the connections
                              * remote context hashcode from the 'ack' message.
                              */
                             final String data =
                                 new String(byteData, ByteConstants.ENCODING).trim();
                             final String[] fields = data.split(DELIMITER);
                             setIdentity(fields[1]);
                             setRemoteContextHashCode(Integer.parseInt(fields[2]));
                             connectionHandshakeComplete();
                             // notify with the new connection
                             getContext().queueEvent(
                                 new ConnectionAvailableEvent(getContext(), this));
                         }
                         else
                         {
                             if (getLog().isWarnEnabled())
                             {
                                 getLog().warn(
                                     "Connection not properly established"
                                         + " so could not process message '"
                                         + new String(byteData,
                                             ByteConstants.ENCODING).trim()
                                         + "'");
                             }
                             // undetermined state, so shutdown
                             destroy();
                         }
                     }
                 }
                 else
                 {
                     final MessageEvent messageEvent =
                         new MessageEvent(this, byteData);
                     messageEvent.setTriggerEvent(new MessageConsumedEvent(this,
                         messageEvent));
                     synchronized (this)
                     {
                         this.messagesToProcess++;
                     }
                     getContext().queueEvent(messageEvent);
                 }
             }
         }
         catch (IOException e)
         {
             logException(getLog(), "Destroying " + this
                 + " because data could not be read from "
                 + getSelectionKey().channel(), e);
             this.destroyWhenMessagesProcessed = true;
             destroy();
         }
     }
 
     /**
      * Writes any data in the message buffer to the socket
      */
     private void write()
     {
         List<ByteBuffer> copy = CollectionFactory.newList(this.messages.size());
         synchronized (this)
         {
             copy.addAll(this.messages);
             this.messages.clear();
         }
         for (Iterator<ByteBuffer> iterator = copy.iterator(); iterator.hasNext();)
         {
             final ByteBuffer data = iterator.next();
             try
             {
                 if (getLog().isTraceEnabled())
                 {
                     getLog().trace(
                         this + " sending '" + Arrays.toString(data.array())
                             + "'");
                 }
                 ((SocketChannel) getSelectionKey().channel()).write(data);
             }
             catch (IOException e)
             {
                 logException(getLog(), this + " could not write data '"
                     + new String(data.array()) + "'", e);
             }
         }
         synchronized (this)
         {
             if (this.messages.size() == 0)
             {
                 // nothing more to write, so remove write interest
                 getSelectionKey().interestOps(
                     getSelectionKey().interestOps() ^ SelectionKey.OP_WRITE);
             }
         }
     }
 
     @Override
     protected void doDestroy()
     {
         super.doDestroy();
         try
         {
             if (getSelectionKey() != null
                 && getSelectionKey().channel() != null)
             {
                 getSelectionKey().channel().close();
             }
         }
         catch (Exception e)
         {
             logException(getLog(), "Could not close channel for " + this, e);
         }
     }
 
     public SelectionKey getSelectionKey()
     {
         return selectionKeyTask.getSelectionKey();
     }
 
     public void setSelectionKey(SelectionKey selectionKey)
     {
         selectionKeyTask.setSelectionKey(selectionKey);
     }
 
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     @Override
     protected void doComponentDestroy()
     {
         super.doComponentDestroy();
         this.broker.getSelectorTasks().unregister(
             this.selectionKeyTask.getSelectionKey());
     }
 }