/*
      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.channel;
  
  import static fulmine.util.Utils.CLOSE_BRACE;
  import static fulmine.util.Utils.CLOSE_CURLY;
  import static fulmine.util.Utils.COMMA_SPACE;
  import static fulmine.util.Utils.OPEN_BRACE;
  import static fulmine.util.Utils.OPEN_CURLY;
  import static fulmine.util.Utils.SPACE;
  import static fulmine.util.Utils.logException;
  import static fulmine.util.Utils.nullCheck;
  import static fulmine.util.Utils.safeToString;
  
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.List;
  import java.util.Map;
  
  import fulmine.AbstractLifeCycle;
  import fulmine.Domain;
  import fulmine.IAddressable;
  import fulmine.IDomain;
  import fulmine.IType;
  import fulmine.Type;
  import fulmine.context.IFrameworkContext;
  import fulmine.context.FulmineContext.SystemInfoFields;
  import fulmine.distribution.connection.IConnection;
  import fulmine.distribution.events.MessageEvent;
  import fulmine.event.IEvent;
  import fulmine.event.IEventSource;
  import fulmine.event.listener.AbstractEventHandler;
  import fulmine.event.listener.IEventListener;
  import fulmine.event.listener.ILifeCycleEventListener;
  import fulmine.event.listener.IPriorityEventListener;
  import fulmine.event.subscription.ISubscriptionManager;
  import fulmine.event.subscription.ISubscriptionParameters;
  import fulmine.event.subscription.SubscriptionParameters;
  import fulmine.event.system.ISystemEventListener;
  import fulmine.event.system.RxEvent;
  import fulmine.event.system.TxEvent;
  import fulmine.model.container.IContainer;
  import fulmine.model.container.IContainer.DataState;
  import fulmine.model.container.events.ContainerDestroyedEvent;
  import fulmine.model.container.events.ContainerStateChangeEvent;
  import fulmine.model.container.events.RemoteContainerCreatedEvent;
  import fulmine.model.container.subscription.ContainerSubscriptionManager;
  import fulmine.model.container.subscription.remote.RemoteContainerSubscriptionManager;
  import fulmine.model.container.subscription.remote.RxSubscription;
  import fulmine.model.container.subscription.remote.RxSubscriptionFactory;
  import fulmine.model.container.subscription.remote.TxSubscription;
  import fulmine.model.container.subscription.remote.TxSubscriptionFactory;
  import fulmine.model.field.LongField;
  import fulmine.protocol.specification.ByteConstants;
  import fulmine.protocol.specification.ByteWriter;
  import fulmine.protocol.specification.IFrameReader;
  import fulmine.rpc.events.RpcInvokeEvent;
  import fulmine.util.array.ArrayUtils;
  import fulmine.util.collection.CollectionFactory;
  import fulmine.util.collection.CollectionUtils;
  import fulmine.util.log.AsyncLog;
  
  /**
   * The default channel implementation. This is thread safe. This implementation
   * is composed of multiple collaborator objects to handle:
   * <ul>
   * <li>subscriptions for remote containers
   * <li>subscriptions for local containers from the remote peer channel
   * <li>messages coming from the remote peer channel - see {@link IChannel} for a
   * description of the possible messages
   * </ul>
   * <p>
   * <h2>Subscription handling</h2>
   * All subscribe and unsubscribe methods are handled by two collaborating
   * {@link ISubscriptionManager} objects; a 'transmit' and 'receive' instance. A
   * channel passes all remote container subscription requests to its remote peer.
   * The remote channel hands this subscription request to its 'transmit'
   * subscription manager. This 'transmit' subscription manager handles
   * registering for {@link TxEvent}s generated by the subscribed container(s);
   * the events are then sent to the peer channel. It also handles situations
   * where the container(s) may not exist yet. The local channel hands the
   * subscription request to its 'receive' subscription manager which handles
   * adding the subscription listener to the remote container. The 'receive'
   * subscription manager also handles situations where the remote container does
   * not exist yet; it listens for {@link RemoteContainerCreatedEvent}.
   * <p>
  * When a channel connects to another channel, they exchange a sequence of
  * messages to synchronise readiness with each other. A channel registers for
  * the following system events:
  * <ul>
  * <li>{@link ContainerDestroyedEvent}
  * </ul>
  * When a channel is notified with this event it transmits it to the remote
  * channel (if there is a subscription) and this allows remote containers to be
  * marked as {@link DataState#STALE} and then destroyed.
  * <p>
  * The diagram below illustrates the subscription mechanics.
  * 
  * <pre>
  *          context       local_channel          remote_channel
  *             |               |                    |
  *             |               |                    |&lt;---- subscribe for
  *             |               |                    |      remote container A
  *             |               |&lt;--- subscribe -----|
  *             |               |                    |
  *             |               |                    |
  *             |               |                    |
  *             |               |                    |
  * container A |               |                    |
  *  created --&gt;|               |                    |
  *             |-- TxEvent ---&gt;|                    |
  *             | container A   |----- TxEvent -----&gt;|
  *             |               |                    |----&gt; create remote 
  *             |               |                    |      container A
  * </pre>
  * 
  * When a channel is destroyed, it sends a {@link #MSG_DESTROY_CONNECTION} to
  * the other channel so the other channel can destroy the connection. This
  * allows messages sent by this channel to be processed by the other end and
  * only after the messages have been processed will the connection be severed.
  * 
  * @author Ramon Servadei
  */
 public class Channel extends AbstractLifeCycle implements IChannel,
     IChannelOperations
 {
     final static AsyncLog LOG = new AsyncLog(Channel.class);
 
     /** The shared state for the channel */
     private final IChannelState state;
 
     /**
      * The standard length in bytes for the header portion of the messages
      * exchanged.
      */
     static final int HEADER_LEN =
         ByteWriter.getBytes(IChannel.MSG_SUBSCRIBE).length;
 
     /** The byte[] for {@link IChannel#MSG_SUBSCRIBE} */
     static final byte[] MSG_SUBSCRIBE_BYTES =
         ByteWriter.getBytes(IChannel.MSG_SUBSCRIBE);
 
     /** The byte[] for {@link IChannel#MSG_UNSUBSCRIBE} */
     static final byte[] MSG_UNSUBSCRIBE_BYTES =
         ByteWriter.getBytes(IChannel.MSG_UNSUBSCRIBE);
 
     /** The byte[] for {@link IChannel#MSG_DATA} */
     static final byte[] MSG_DATA_BYTES = ByteWriter.getBytes(IChannel.MSG_DATA);
 
     /** The byte[] for {@link IChannel#MSG_CONTAINER_DESTROYED} */
     static final byte[] MSG_CONTAINER_DESTROYED_BYTES =
         ByteWriter.getBytes(IChannel.MSG_CONTAINER_DESTROYED);
 
     /** The byte[] for {@link IChannel#MSG_SYN} */
     static final byte[] MSG_SYN_BYTES = ByteWriter.getBytes(IChannel.MSG_SYN);
 
     /** The byte[] for {@link IChannel#MSG_SYN_ACK} */
     static final byte[] MSG_SYN_ACK_BYTES =
         ByteWriter.getBytes(IChannel.MSG_SYN_ACK);
 
     /** The byte[] for {@link IChannel#MSG_RETRANSMIT} */
     static final byte[] MSG_RETRANSMIT_BYTES =
         ByteWriter.getBytes(IChannel.MSG_RETRANSMIT);
 
     /** The byte[] for {@link IChannel#MSG_RETRANSMIT_ALL} */
     static final byte[] MSG_RETRANSMIT_ALL_BYTES =
         ByteWriter.getBytes(IChannel.MSG_RETRANSMIT_ALL);
 
     /** The byte[] for {@link IChannel#MSG_DESTROY_CONNECTION} */
     static final byte[] MSG_DESTROY_CONNECTION_BYTES =
         ByteWriter.getBytes(IChannel.MSG_DESTROY_CONNECTION);
 
     /** The byte[] for {@link IChannel#MSG_INVOKE_RPC} */
     static final byte[] MSG_INVOKE_RPC_BYTES =
         ByteWriter.getBytes(IChannel.MSG_INVOKE_RPC);
 
     /**
      * Utility method to create a {@link IChannel#MSG_DATA} message using the
      * data
      * 
      * @param data
      *            that data for the data message body
      * @return the complete data message structure to send to a peer channel
      */
     final static byte[] createDataMessage(byte[] data)
     {
         // prepend the update message header to the data
         final int headerLen = Channel.MSG_DATA_BYTES.length;
         byte[] message = new byte[data.length + headerLen];
         System.arraycopy(data, 0, message, headerLen, data.length);
         System.arraycopy(Channel.MSG_DATA_BYTES, 0, message, 0, headerLen);
         return message;
     }
 
     /**
      * Construct the channel
      * 
      * @param connection
      *            the connection to the remote context
      * @param context
      *            the local context for the channel
      */
     public Channel(IConnection connection, IFrameworkContext context)
     {
         /*
          * NOTE: TxEvents and RxEvents are bound to containers, not the event
          * type (unlike the container created/destroyed events). TxEvent and
          * RxEvents are directed to the ChannelState's eventListener by code in
          * the channel.
          */
         this(new ChannelState(context, connection));
     }
 
     /**
      * Internally chained constructor with the state to use
      * 
      * @param state
      *            the state for the channel
      */
     protected Channel(IChannelState state)
     {
         super();
         this.state = state;
     }
 
     public final boolean subscribe(ISubscriptionParameters parameters)
     {
         checkActive();
         if (getLog().isDebugEnabled())
         {
             getLog().debug(this + "Subscribe for " + safeToString(parameters));
         }
         final boolean subscribed =
             getState().getRxSubscriptionManager().subscribe(parameters);
         // always try to re-subscribe - all these ops are idempotent anyway
         doSubscribeOperation(parameters);
         send(ByteWriter.getBytes(Channel.MSG_SUBSCRIBE
             + parameters.getType().value() + Channel.DELIMITER
             + parameters.getDomain().value() + Channel.DELIMITER
             + parameters.getIdentity()));
         return subscribed;
     }
 
     /**
      * Perform any operation required when a subscription occurs.
      * 
      * @param parameters
      *            the parameters identifying the subscription
      */
     protected void doSubscribeOperation(ISubscriptionParameters parameters)
     {
         // direct any RxEvents to the ChannelTransmissionListener
         getState().getRxSubscriptionManager().addListener(parameters,
             getState().getEventHandler());
     }
 
     /**
      * if there are no requests from the remote context, we can destroy the
      * channel.
      */
     private void checkIfConnectionCanBeClosed()
     {
         if (getState().getTxSubscriptionManager().getSubscribedSources().size() == 0
             && getState().getRxSubscriptionManager().getSubscribedSources().size() == 0)
         {
             if (getLog().isInfoEnabled())
             {
                 getLog().info(
                     "No RX or TX subscriptions so destroying "
                         + safeToString(this));
             }
             destroy();
         }
     }
 
     public final boolean unsubscribe(ISubscriptionParameters parameters)
     {
         checkActive();
         // this is only called by the distribution manager when there are no
         // listeners registered against the subscription parameters when
         // processing an unsubscription
         if (getLog().isDebugEnabled())
         {
             getLog().debug(this + "Unsubscribe for " + safeToString(parameters));
         }
         final boolean unsubscribed =
             getState().getRxSubscriptionManager().unsubscribe(parameters);
         doUnsubscribeOperation(parameters);
         // always send the unsubscribe to the remote context, regardless
         send(ByteWriter.getBytes(Channel.MSG_UNSUBSCRIBE
             + parameters.getType().value() + Channel.DELIMITER
             + parameters.getDomain().value() + Channel.DELIMITER
             + parameters.getIdentity()));
         checkIfConnectionCanBeClosed();
         return unsubscribed;
     }
 
     /**
      * Perform any operation required when an unsubscription occurs.
      * 
      * @param parameters
      *            the parameters identifying subscription
      */
     protected void doUnsubscribeOperation(ISubscriptionParameters parameters)
     {
         // noop
     }
 
     public final void send(byte[] bytes)
     {
         checkActive();
         getState().getConnection().send(bytes);
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     @Override
     protected void doStart()
     {
         getState().init(this, this);
         getState().start();
         if (getLog().isDebugEnabled())
         {
             getLog().debug(this + " sending SYN");
         }
         send(Channel.MSG_SYN_BYTES);
     }
 
     @Override
     protected final void doDestroy()
     {
         if (getLog().isInfoEnabled())
         {
             getLog().info("Destroying " + this + " (stacktrace follows)",
                 new Exception());
         }
         // process the remote subscriptions
         final Collection<? extends IEventSource> rxSubscribedSources =
             getState().getRxSubscriptionManager().getSubscribedSources();
         if (getLog().isTraceEnabled())
         {
             getLog().trace(
                 this
                     + "[channel destroy] RX subscribed sources to process are "
                     + CollectionUtils.toFormattedString(rxSubscribedSources));
         }
         final int size = rxSubscribedSources.size();
         int count = 0;
         for (IEventSource eventSource : rxSubscribedSources)
         {
             try
             {
                 doChannelDestroyedOperation(eventSource, ++count == size);
             }
             catch (Exception e)
             {
                 logException(getLog(), eventSource, e);
             }
         }
         // stop receiving any messages from the remote context (via the
         // connection)
         getState().getConnection().removeListener(getState().getEventHandler());
         // all TX subscriptions are handled by the TxManager destroy
         getState().destroy();
     }
 
     /**
      * Perform the operation required on the remote {@link IEventSource} that
      * originated from the remote context this channel connects to, when the
      * channel is destroyed.
      * 
      * @param eventSource
      *            a remote event source that is associated with the remote
      *            context this channel connects to
      * @param last
      *            signals if this is the last event source
      */
     protected void doChannelDestroyedOperation(IEventSource eventSource,
         boolean last)
     {
         eventSource.destroy();
     }
 
     public final String getRemoteContextIdentity()
     {
         return getState().getConnection().getRemoteContextIdentity();
     }
 
     public final String toString()
     {
         return getToString(getConnection());
     }
 
     public final boolean isConnectionSyn()
     {
         return getState().isConnectionSyn();
     }
 
     public final void setConnectionSyn()
     {
         getState().setConnectionSyn();
     }
 
     public final void requestRetransmit(String identityRegularExpression,
         IType type, IDomain domain)
     {
         checkActive();
         try
         {
             send(ByteWriter.getBytes(Channel.MSG_RETRANSMIT + type.value()
                 + Channel.DELIMITER + domain.value() + Channel.DELIMITER
                 + identityRegularExpression));
         }
         catch (Exception e)
         {
             logException(getLog(), this + " idregex="
                 + identityRegularExpression + COMMA_SPACE + type + COMMA_SPACE
                 + domain, e);
         }
     }
 
     public final void requestRetransmitAll()
     {
         checkActive();
         send(Channel.MSG_RETRANSMIT_ALL_BYTES);
     }
 
     public final void retransmitAll()
     {
         checkActive();
         if (getLog().isDebugEnabled())
         {
             getLog().debug(this + " retransmitting all subscribed containers");
         }
         final Collection<? extends IEventSource> subscribedContainers =
             getState().getTxSubscriptionManager().getSubscribedSources();
         doRetransmit(subscribedContainers);
     }
 
     public final void retransmit(String identityRegularExpression, IType type,
         IDomain domain)
     {
         checkActive();
         if (getLog().isDebugEnabled())
         {
             getLog().debug(
                 this + " retransmitting containers for identityPattern='"
                     + identityRegularExpression + "'" + ", type=" + type
                     + ", domain=" + domain);
         }
         final Collection<IEventSource> subscribedContainers =
             getState().getTxSubscriptionManager().getSubscribedSources(
                 new SubscriptionParameters(identityRegularExpression, type,
                     domain));
         if (subscribedContainers.size() == 0)
         {
             // try a single container
             if (getState().getContext().containsLocalContainer(
                 identityRegularExpression, type, domain))
             {
                 subscribedContainers.add(getState().getContext().getLocalContainer(
                     identityRegularExpression, type, domain));
             }
         }
         doRetransmit(subscribedContainers);
     }
 
     void doRetransmit(
         final Collection<? extends IEventSource> subscribedContainers)
     {
         getState().getContext().execute(new Runnable()
         {
             public void run()
             {
                 final long start = System.nanoTime();
                 for (IEventSource container : subscribedContainers)
                 {
                     try
                     {
                         // Note: this may block if there is a frame waiting and
                         // the container is bound to a different event
                         // processor.
                         send(Channel.createDataMessage(Channel.this.getState().getContext().getFrameWriter().writeComplete(
                             (IContainer) container)));
                     }
                     catch (Exception e)
                     {
                         logException(getLog(), container, e);
                     }
                 }
                 final long sendTime = (System.nanoTime() - start);
                 if (Channel.LOG.isDebugEnabled())
                 {
                     Channel.LOG.debug("Time to retransmit "
                         + subscribedContainers.size() + " containers: "
                         + sendTime + " nanoseconds");
                 }
             }
         });
     }
 
     public final IConnection getConnection()
     {
         return getState().getConnection();
     }
 
     private String getToString(IConnection connection)
     {
         return getClass().getSimpleName() + SPACE + OPEN_BRACE
             + (isActive() ? "active" : "inactive") + CLOSE_BRACE + SPACE
             + OPEN_CURLY + connection + OPEN_BRACE
             + connection.getRemoteContextHashCode() + CLOSE_BRACE + CLOSE_CURLY
             + SPACE;
     }
 
     /** Holds the shared state of the channel */
     IChannelState getState()
     {
         return this.state;
     }
 
     public final boolean addListener(ISubscriptionParameters parameters,
         IEventListener listener)
     {
         return getState().getRxSubscriptionManager().addListener(parameters,
             listener);
     }
 
     public final void eventSourceCreated(IAddressable identity)
     {
         getState().getRxSubscriptionManager().eventSourceCreated(identity);
     }
 
     public final void eventSourceDestroyed(IAddressable identity)
     {
         getState().getRxSubscriptionManager().eventSourceDestroyed(identity);
     }
 
     public final Collection<IEventSource> getSubscribedSources()
     {
         return getState().getRxSubscriptionManager().getSubscribedSources();
     }
 
     public final Collection<IEventSource> getSubscribedSources(
         ISubscriptionParameters parameters)
     {
         return getState().getRxSubscriptionManager().getSubscribedSources(
             parameters);
     }
 
     public final boolean isSubscribed(IEventSource source)
     {
         return getState().getRxSubscriptionManager().isSubscribed(source);
     }
 
     public final boolean removeListener(ISubscriptionParameters parameters,
         IEventListener listener)
     {
         return getState().getRxSubscriptionManager().removeListener(parameters,
             listener);
     }
 
     public final boolean includes(IAddressable parameters)
     {
         return getState().getRxSubscriptionManager().includes(parameters);
     }
 
     public List<IEventListener> getListeners(ISubscriptionParameters parameters)
     {
         return getState().getRxSubscriptionManager().getListeners(parameters);
     }
 
     public final byte[] getContainerDestroyedMessage(
         ContainerDestroyedEvent event)
     {
         return ByteWriter.getBytes(Channel.MSG_CONTAINER_DESTROYED
             + event.getType().value() + Channel.DELIMITER
             + event.getDomain().value() + Channel.DELIMITER
             + event.getIdentity());
     }
 
     public void invokeRpc(String remoteContextIdentity, byte[] rpcData)
     {
         send(ArrayUtils.merge(ByteWriter.getBytes(Channel.MSG_INVOKE_RPC),
             rpcData));
     }
 }
 
 /**
  * The internal state for a {@link Channel}.
  * 
  * @author Ramon Servadei
  * 
  */
 class ChannelState extends AbstractLifeCycle implements IChannelState
 {
     private final static AsyncLog LOG = new AsyncLog(ChannelState.class);
 
     private final IFrameworkContext context;
 
     /**
      * The subscription manager for {@link TxSubscription} instances. This
      * manager handles subscriptions for transmitting local containers to the
      * remote context; these are subscriptions from the remote context for local
      * containers.
      */
     private final ISubscriptionManager txSubscriptionManager;
 
     /**
      * The subscription manager for {@link RxSubscription} instances. This
      * manager handles subscriptions for receiving remote containers from remote
      * contexts; these are subscriptions from the local context for remote
      * containers.
      */
     private final ISubscriptionManager rxSubscriptionManager;
 
     private boolean connectionSyn;
 
     private final IConnection connection;
 
     private ILifeCycleEventListener eventHandler;
 
     /** Fields for tracking TX and RX counts */
     private final LongField txCount, rxCount;
 
     /**
      * Holds the RxEvent listener token against the application subscription
      * token
      */
     private final Map<Integer, Integer> tokenMap;
 
     ChannelState(IFrameworkContext context, IConnection connection)
     {
         super();
         this.context = context;
         this.connection = connection;
         // this is not a remote subscription manager because it transmits local
         // containers to the remote channel
         this.txSubscriptionManager =
             new ContainerSubscriptionManager(context,
                 new TxSubscriptionFactory());
         this.rxSubscriptionManager =
             new RemoteContainerSubscriptionManager(context,
                 connection.getRemoteContextIdentity(),
                 new RxSubscriptionFactory());
         this.tokenMap = CollectionFactory.newMap();
         this.txCount =
             new LongField(getConnection().getRemoteContextIdentity() + ":"
                 + SystemInfoFields.TX_COUNT);
         this.rxCount =
             new LongField(getConnection().getRemoteContextIdentity() + ":"
                 + SystemInfoFields.RX_COUNT);
         getContext().getSystemInfo().add(ChannelState.this.txCount);
         getContext().getSystemInfo().add(ChannelState.this.rxCount);
     }
 
     public Map<Integer, Integer> getTokenMap()
     {
         return this.tokenMap;
     }
 
     /**
      * Initialise the state
      * 
      * @param channel
      * @param channelOps
      */
     @SuppressWarnings("unchecked")
     public final void init(IChannel channel, IChannelOperations channelOps)
     {
         Map<Class<? extends IEvent>, IEventListener> listeners =
             AbstractEventHandler.getEventHandlerMappings(createEventHandlers(
                 context, channel, channelOps));
         this.eventHandler =
             new ChannelTransmissionListener(OPEN_CURLY
                 + Channel.class.getSimpleName() + " to "
                 + connection.toString() + CLOSE_CURLY, context, listeners);
 
         // get ready to process messages from the peer channel
         getConnection().addListener(getEventHandler());
     }
 
     /**
      * Create the event handlers that will react to the events and system events
      * that the channel should handle.
      * 
      * @param context
      * @param channel
      * @param channelOps
      * @return an array of {@link AbstractEventHandler}
      */
     @SuppressWarnings("unchecked")
     AbstractEventHandler[] createEventHandlers(IFrameworkContext context,
         IChannel channel, IChannelOperations channelOps)
     {
         return new AbstractEventHandler[] {
             new MessageEventHandler(channel, channelOps, this),
             new RxEventHandler(channelOps, this),
             new ContainerDestroyedEventHandler(channelOps, this),
             new ContainerStateChangeEventHandler(channelOps, this),
             new TxEventHandler(channelOps, this), };
     }
 
     protected final void doDestroy()
     {
         // signal the other end to break the connection
         if (getConnection().isActive())
         {
             try
             {
                 getConnection().send(Channel.MSG_DESTROY_CONNECTION_BYTES);
             }
             catch (Exception e)
             {
                 logException(getLog(), this, e);
             }
         }
         getContext().getSystemInfo().remove(this.txCount);
         getContext().getSystemInfo().remove(this.rxCount);
         doComponentDestroy();
     }
 
     /**
      * Delegate method to perform the destroy operation on this component.
      */
     void doComponentDestroy()
     {
         this.txSubscriptionManager.destroy();
         this.rxSubscriptionManager.destroy();
         this.eventHandler.destroy();
     }
 
     protected void doStart()
     {
         this.txSubscriptionManager.start();
         this.rxSubscriptionManager.start();
         this.eventHandler.start();
         // chances are the connection is already started
         this.connection.start();
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     /**
      * Helper to process events received from the peer channel and from the
      * local context.
      */
     public final ILifeCycleEventListener getEventHandler()
     {
         return eventHandler;
     }
 
     public final void setConnectionSyn()
     {
         this.connectionSyn = true;
     }
 
     /** The local context */
     public final IFrameworkContext getContext()
     {
         return this.context;
     }
 
     /**
      * The subscription manager for {@link TxSubscription} instances. This
      * manager handles subscriptions for transmitting local containers to the
      * remote context; these are subscriptions from the remote context for local
      * containers.
      */
     public final ISubscriptionManager getTxSubscriptionManager()
     {
         return this.txSubscriptionManager;
     }
 
     /**
      * The subscription manager for {@link RxSubscription} instances. This
      * manager handles subscriptions for receiving remote containers from remote
      * contexts; these are subscriptions from the local context for remote
      * containers.
      */
     public final ISubscriptionManager getRxSubscriptionManager()
     {
         return this.rxSubscriptionManager;
     }
 
     /**
      * Indicates if the connection to the peer {@link IChannel} has been
      * synchronised. The peer channels in the peer contexts will be created at
      * different times, so they must synchronise with each other to ensure no
      * missed registrations occur.
      */
     public final boolean isConnectionSyn()
     {
         return this.connectionSyn;
     }
 
     /** The connection to use */
     public final IConnection getConnection()
     {
         return this.connection;
     }
 
     public final LongField getTxCount()
     {
         return this.txCount;
     }
 
     public final LongField getRxCount()
     {
         return this.rxCount;
     }
 }
 
 /**
  * Base class for channel event handlers.
  * <p>
  * All sub-classes are thread safe.
  * 
  * @author Ramon Servadei
  * @param <EVENT>
  *            the event type the handler works with
  */
 abstract class ChannelEventHandler<EVENT extends IEvent> extends
     AbstractEventHandler<EVENT> implements ISystemEventListener
 {
     /** The channel operations for the handler */
     private final IChannelOperations channelOps;
 
     /** The channel state for the handler */
     private final IChannelState state;
 
     ChannelEventHandler(IChannelOperations channelOps, IChannelState state)
     {
         super();
         nullCheck(channelOps, "No channel operations provided");
         nullCheck(state, "No state provided");
         this.channelOps = channelOps;
         this.state = state;
     }
 
     IChannelState getState()
     {
         return state;
     }
 
     IChannelOperations getChannelOps()
     {
         return channelOps;
     }
 }
 
 /**
  * Handles {@link ContainerStateChangeEvent}s and sends them to any remote
  * contexts that have subscribed for the identified container.
  * 
  * @author Ramon Servadei
  */
 final class ContainerStateChangeEventHandler extends
     ChannelEventHandler<ContainerStateChangeEvent>
 {
 
     private final static AsyncLog LOG =
         new AsyncLog(ContainerStateChangeEventHandler.class);
 
     ContainerStateChangeEventHandler(IChannelOperations channelOps,
         IChannelState state)
     {
         super(channelOps, state);
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     @Override
     public void handle(ContainerStateChangeEvent event)
     {
         if (getState().getTxSubscriptionManager().isSubscribed(
             event.getContainer()))
         {
             getChannelOps().send(
                 Channel.createDataMessage(getState().getContext().getFrameWriter().writeMeta(
                     event.getContainer())));
         }
     }
 
 }
 
 /**
  * Handler for {@link RxEvent}s. Extracts the data from the {@link RxEvent} and
  * uses an {@link IFrameReader} to apply the data to the correct
  * {@link IContainer}. This is the class that updates remote containers.
  * 
  * @author Ramon Servadei
  * 
  */
 final class RxEventHandler extends ChannelEventHandler<RxEvent>
 {
 
     private final static AsyncLog LOG = new AsyncLog(RxEventHandler.class);
 
     /** counter for the RX events */
     private long count;
 
     RxEventHandler(IChannelOperations channelOps, IChannelState state)
     {
         super(channelOps, state);
     }
 
     /**
      * Extracts the data from the {@link RxEvent} and uses an
      * {@link IFrameReader} to apply the data to the correct {@link IContainer}.
      */
     @Override
     public void handle(RxEvent event)
     {
         getState().getRxCount().set(count++);
         if (getLog().isTraceEnabled())
         {
             getLog().trace(event.getTraceString());
         }
         // this ensures the container's event processor thread runs this
         getState().getContext().getFrameReader().read(event.getBuffer(),
             getState().getConnection().getRemoteContextIdentity(),
             getState().getContext());
     }
 
     @Override
     public AsyncLog getLog()
     {
         return LOG;
     }
 }
 
 /**
  * Handler for {@link TxEvent}s. Extracts the data from the {@link TxEvent} and
  * sends it to the peer channel.
  * 
  * @author Ramon Servadei
  * 
  */
 final class TxEventHandler extends ChannelEventHandler<TxEvent>
 {
 
     private final static AsyncLog LOG = new AsyncLog(TxEventHandler.class);
 
     /** counter for the TX events */
     private long count;
 
     TxEventHandler(IChannelOperations channelOps, IChannelState state)
     {
         super(channelOps, state);
     }
 
     /**
      * Extracts the data from the {@link TxEvent} and sends it to the peer
      * channel.
      */
     @Override
     public void handle(TxEvent event)
     {
         getState().getTxCount().set(count++);
         if (getLog().isTraceEnabled())
         {
             getLog().trace(event.getTraceString());
         }
         // if the source of the event has been destroyed, don't send
         if (getState().getContext().containsLocalContainer(
             event.getSource().getIdentity(), event.getSource().getType(),
             event.getSource().getDomain()))
         {
             byte[] data = event.getBuffer();
             getChannelOps().send(Channel.createDataMessage(data));
         }
         else
         {
             if (getLog().isDebugEnabled())
             {
                 getLog().debug("Source is destroyed, not sending " + event);
             }
         }
     }
 
     @Override
     public AsyncLog getLog()
     {
         return LOG;
     }
 }
 
 /**
  * Handler for {@link ContainerDestroyedEvent}s. Passes the event to the peer
  * channel.
  * 
  * @author Ramon Servadei
  * 
  */
 final class ContainerDestroyedEventHandler extends
     ChannelEventHandler<ContainerDestroyedEvent> implements
     IPriorityEventListener
 {
 
     private final static AsyncLog LOG =
         new AsyncLog(ContainerDestroyedEventHandler.class);
 
     ContainerDestroyedEventHandler(IChannelOperations channelOps,
         IChannelState state)
     {
         super(channelOps, state);
     }
 
     /**
      * Passes the event to the peer channel.The peer channel's
      * {@link MessageEventHandler#doRemoteContainerDestroyed(String, IType, IDomain)}
      * will receive this.
      */
     @Override
     public void handle(ContainerDestroyedEvent event)
     {
         if (getState().getTxSubscriptionManager().includes(
             new SubscriptionParameters(event.getIdentity(), event.getType(),
                 event.getDomain())))
         {
             getChannelOps().send(
                 getChannelOps().getContainerDestroyedMessage(event));
         }
     }
 
     @Override
     public AsyncLog getLog()
     {
         return LOG;
     }
 }
 
 /**
  * A handler for {@link MessageEvent}s. The message event payload is one of the
  * messages declared in the {@link IChannel} so this class provides the logic to
  * handle each type of message. Generally, each message is handled by a doXXX
  * method.
  * 
  * @author Ramon Servadei
  * 
  */
 class MessageEventHandler extends ChannelEventHandler<MessageEvent>
 {
 
     private final static AsyncLog LOG = new AsyncLog(MessageEventHandler.class);
 
     /** The channel for this handler */
     private final IChannel channel;
 
     /**
      * Standard constructor
      * 
      * @param channel
      *            the channel
      * @param channelOps
      *            the channel operations
      * @param state
      *            the channel state
      */
     MessageEventHandler(IChannel channel, IChannelOperations channelOps,
         IChannelState state)
     {
         super(channelOps, state);
         this.channel = channel;
     }
 
     @Override
     public void handle(MessageEvent event)
     {
         if (getLog().isTraceEnabled())
         {
             getLog().trace(event.getTraceString());
         }
         final byte[] data = event.getData();
         byte[] arg = new byte[data.length - Channel.HEADER_LEN];
         System.arraycopy(data, Channel.HEADER_LEN, arg, 0, arg.length);
 
         if (ArrayUtils.startsWith(Channel.MSG_DATA_BYTES, data))
         {
             doMessageReceived(arg);
             return;
         }
         if (ArrayUtils.startsWith(Channel.MSG_INVOKE_RPC_BYTES, data))
         {
             doInvokeRpc(arg);
             return;
         }
         try
         {
             final String payload = new String(arg, ByteConstants.ENCODING);
             final String[] elements = splitTypeDomainAndIdentity(payload);
             if (ArrayUtils.startsWith(Channel.MSG_SUBSCRIBE_BYTES, data))
             {
                 // subscription request from peer
                 doTxSubscribe(elements[2],
                     Type.get(Byte.parseByte(elements[0])),
                     Domain.get(Byte.parseByte(elements[1])));
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_UNSUBSCRIBE_BYTES, data))
             {
                 // unsubscribe request from peer
                 doTxUnsubscribe(elements[2],
                     Type.get(Byte.parseByte(elements[0])),
                     Domain.get(Byte.parseByte(elements[1])));
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_CONTAINER_DESTROYED_BYTES,
                 data))
             {
                 // a container destroyed in the peer
                 doRemoteContainerDestroyed(elements[2],
                     Type.get(Byte.parseByte(elements[0])),
                     Domain.get(Byte.parseByte(elements[1])));
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_SYN_BYTES, data))
             {
                 // a syn from the peer
                 doSynReceived();
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_SYN_ACK_BYTES, data))
             {
                 // an ack from the peer
                 doSynAckReceived();
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_RETRANSMIT_BYTES, data))
             {
                 doRetransmit(elements[2],
                     Type.get(Byte.parseByte(elements[0])),
                     Domain.get(Byte.parseByte(elements[1])));
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_RETRANSMIT_ALL_BYTES, data))
             {
                 doRetransmitAll();
                 return;
             }
             if (ArrayUtils.startsWith(Channel.MSG_DESTROY_CONNECTION_BYTES,
                 data))
             {
                 doDestroyConnection();
                 return;
             }
             throw new RuntimeException("Unhandled message=" + payload);
         }
         catch (Exception e)
         {
             throw new RuntimeException(
                 "Could not handle message (unencoded): '"
                     + Arrays.toString(data) + "'", e);
         }
     }
 
     /**
      * Handle an RPC invoke event. This is encapsulated in an
      * {@link RpcInvokeEvent} and fired at the context for it to locate the
      * appropriate handler
      * 
      * @param rpcData
      *            the RPC data
      */
     void doInvokeRpc(byte[] rpcData)
     {
         RpcInvokeEvent event =
             new RpcInvokeEvent(getState().getContext(), rpcData);
         // the RPC manager should pick this up
         getState().getContext().queueEvent(event);
     }
 
     /**
      * Handle a data message. This is generally an update to a remote container.
      * 
      * @param arg
      *            the byte[] encapsulating the data message
      */
     void doMessageReceived(byte[] arg)
     {
         // data for a remote container from the peer
         // NOTE: this will raise a ContainerCreatedEvent if it creates the
         // container
         final IContainer remoteContainer =
             getState().getContext().getFrameReader().getRemoteContainerForFrame(
                 arg, getState().getConnection().getRemoteContextIdentity(),
                 getState().getContext());
         /*
          * encapsulate in an RxEvent so the container's event processor will
          * handle this - spreads the load across all event processors (depending
          * on how the containers are allocated) rather than a single thread
          * executing all remote container events
          */
         RxEvent rx = new RxEvent(remoteContainer, arg);
         // add the receive event to the context
         getState().getContext().queueEvent(rx);
     }
 
     /**
      * Handles the signal that the identified remote container has been
      * destroyed in its local context. The remote container instance in this
      * context needs to be destroyed.
      * 
      * @param identityRegex
      *            remote container identity
      * @param type
      *            remote container type
      * @param domain
      *            remote container domain
      */
     void doRemoteContainerDestroyed(String identity, IType type, IDomain domain)
     {
         final IContainer remoteContainer =
             getState().getContext().getRemoteContainer(
                 getState().getConnection().getRemoteContextIdentity(),
                 identity, type, domain);
         remoteContainer.destroy();
     }
 
     final void doSynAckReceived()
     {
         if (!getChannelOps().isConnectionSyn())
         {
             if (Channel.LOG.isDebugEnabled())
             {
                 Channel.LOG.debug(getChannel()
                     + " received SYN_ACK, channel is ready.");
             }
             getChannelOps().setConnectionSyn();
             ChannelReadyEvent event =
                 new ChannelReadyEvent(getState().getContext(), getChannel());
             getState().getContext().queueEvent(event);
         }
         else
         {
             if (getLog().isDebugEnabled())
             {
                 getLog().debug(
                     getChannel()
                         + " is already ready, ignoring duplicate SYN_ACK.");
             }
         }
     }
 
     final void doSynReceived()
     {
         // always send a SYN_ACK response, even if we are synchronised
         if (Channel.LOG.isDebugEnabled())
         {
             Channel.LOG.debug(getChannel() + " received SYN, sending SYN_ACK");
         }
         getChannelOps().send(Channel.MSG_SYN_ACK_BYTES);
         if (!getChannelOps().isConnectionSyn())
         {
             // The other channel may not have received our first SYN so,
             // for safety, re-transmit the SYN. This might cause the peer
             // channel to retransmit a second SYN_ACK which we will receive -
             // and ignore as it will be a duplicate.
             if (Channel.LOG.isDebugEnabled())
             {
                 Channel.LOG.debug(getChannel() + " re-sending SYN");
             }
             getChannelOps().send(Channel.MSG_SYN_BYTES);
         }
     }
 
     /**
      * Mark matching containers to transmit. TxEvents will be generated by the
      * container(s) and received by the channel's event handler. These are
      * subsequently sent to the remote end of this channel.
      * 
      * @param identityRegex
      *            container identity
      * @param type
      *            container type
      * @param domain
      *            container domain
      */
     void doTxSubscribe(String identityRegex, IType type, IDomain domain)
     {
         /*
          * Creates a TxSubscription that does the business of marking the
          * containers to transmit. This allows the channel to intercept the
          * TxEvents (see update method)
          */
         final SubscriptionParameters parameters =
             new SubscriptionParameters(identityRegex, type, domain);
         if (getLog().isInfoEnabled())
         {
             getLog().info("[subscribe] " + parameters);
         }
         // these operations are idempotent, so even though we may receive
         // duplicate subscriptions, we never handle the duplicates.
         getState().getTxSubscriptionManager().subscribe(parameters);
         getState().getTxSubscriptionManager().addListener(parameters,
             getState().getEventHandler());
     }
 
     /**
      * Unmark matching containers from transmission.
      * 
      * @param identityRegex
      *            container identity to match
      * @param type
      *            container type to match
      * @param domain
      *            container domain to match
      */
     void doTxUnsubscribe(String identityRegex, IType type, IDomain domain)
     {
         // The channel will no longer receive the TxEvents
         final SubscriptionParameters parameters =
             new SubscriptionParameters(identityRegex, type, domain);
         if (getLog().isInfoEnabled())
         {
             getLog().info("[unsubscribe] " + parameters);
         }
         getState().getTxSubscriptionManager().unsubscribe(parameters);
     }
 
     /**
      * Handle the request to retransmit the identified local container
      * 
      * @param identityRegex
      *            container identity to match
      * @param type
      *            container type to match
      * @param domain
      *            container domain to match
      */
     void doRetransmit(String identityRegex, IType type, IDomain domain)
     {
         getChannel().retransmit(identityRegex, type, domain);
     }
 
     /**
      * Handle the request to retransmit all local containers
      */
     void doRetransmitAll()
     {
         getChannel().retransmitAll();
     }
 
     /**
      * Handle the request to destroy the connection
      */
     void doDestroyConnection()
     {
         if (getLog().isInfoEnabled())
         {
             getLog().info(
                 "Received destroy connection message for " + getChannel());
         }
         final IConnection connection = getChannel().getConnection();
         if (connection != null)
         {
             connection.destroy();
         }
     }
 
     /**
      * Utility to split the payload on the '|'
      * 
      * @param payload
      *            the payload containing the type, domain and identity of a
      *            container
      * @return a 3 element String[]
      */
     String[] splitTypeDomainAndIdentity(String payload)
     {
         String[] split = new String[3];
         final int indexOf = payload.indexOf(Channel.DELIMITER);
         if (indexOf > -1)
         {
             split[0] = payload.substring(0, indexOf);
             final int index2 =
                 payload.indexOf(Channel.DELIMITER, indexOf
                     + Channel.DELIMITER.length());
             split[1] =
                 payload.substring(indexOf + Channel.DELIMITER.length(), index2);
             split[2] = payload.substring(index2 + Channel.DELIMITER.length());
         }
         return split;
     }
 
     @Override
     public AsyncLog getLog()
     {
         return LOG;
     }
 
     IChannel getChannel()
     {
         return channel;
     }
 }