/*
      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.context;
  
  import static fulmine.util.Utils.CLOSE_BRACE;
  import static fulmine.util.Utils.COLON;
  import static fulmine.util.Utils.OPEN_BRACE;
  import static fulmine.util.Utils.logException;
  import static fulmine.util.Utils.nullCheck;
  
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  import java.util.Collection;
  import java.util.TimerTask;
  
  import org.apache.commons.lang.builder.HashCodeBuilder;
  
  import fulmine.AbstractLifeCycle;
  import fulmine.Domain;
  import fulmine.IDomain;
  import fulmine.IType;
  import fulmine.Type;
  import fulmine.distribution.IDistributionManager;
  import fulmine.distribution.channel.IChannel;
  import fulmine.distribution.connection.IConnectionBroker;
  import fulmine.distribution.connection.IConnectionDiscoverer;
  import fulmine.event.EventFrameExecution;
  import fulmine.event.EventQueueItem;
  import fulmine.event.IEvent;
  import fulmine.event.IEventManager;
  import fulmine.event.listener.IEventListener;
  import fulmine.event.subscription.ISubscriptionListener;
  import fulmine.event.system.ISystemEvent;
  import fulmine.event.system.ISystemEventSource;
  import fulmine.model.IModelManager;
  import fulmine.model.container.IContainer;
  import fulmine.model.container.IContainerFactory;
  import fulmine.model.field.IField;
  import fulmine.model.field.LongField;
  import fulmine.protocol.specification.IFrameReader;
  import fulmine.protocol.specification.IFrameWriter;
  import fulmine.rpc.IRpcDefinition;
  import fulmine.rpc.IRpcHandler;
  import fulmine.rpc.IRpcManager;
  import fulmine.rpc.IRpcMarker;
  import fulmine.rpc.IRpcPublicationListener;
  import fulmine.rpc.IRpcResult;
  import fulmine.rpc.IRpcResultHandler;
  import fulmine.util.concurrent.ThreadUtils;
  import fulmine.util.log.AsyncLog;
  
  /**
   * The default {@link IFrameworkContext} implementation composed of
   * collaborating implementations:
   * <ul>
   * <li>{@link IModelManager}
   * <li>{@link IEventManager}
   * <li>{@link IDistributionManager}
   * <li>{@link IRpcManager}
   * </ul>
   * <p>
   * This is created with a {@link DefaultPermissionProfile}. By default, all
   * remote container updates are vetoed using an internal
   * {@link VetoingRemoteUpdateHandler}, this can be replaced using an appropriate
   * handler via {@link #setRemoteUpdateHandler(IRemoteUpdateHandler)}.
   * 
   * @author Ramon Servadei
   */
  public final class FulmineContext extends AbstractLifeCycle implements
      IFrameworkContext
  {
  
      /**
       * Encapsulates the field names for the system info record
       * 
       * @author Ramon Servadei
       */
      public static interface SystemInfoFields
      {
          /** Key name for the tx counts in the system info record */
          String TX_COUNT = "tx count";
  
          /** Key name for the rx counts in the system info record */
          String RX_COUNT = "rx count";
  
          /** Key name for the free memory field in the system info record */
         String FREE_MEMORY = "free mem (k)";
 
         /** Key name for the total memory field in the system info record */
         String TOTAL_MEMORY = "total mem (k)";
     }
 
     private final static AsyncLog LOG = new AsyncLog(FulmineContext.class);
 
     static
     {
         try
         {
             // force a class load to set framework types
             Class.forName(Type.class.getName());
             Class.forName(Domain.class.getName());
             ThreadUtils.startThreadDumps();
         }
         catch (Exception e)
         {
             LOG.fatal("Could not load required class", e);
             System.exit(2);
         }
     }
 
     /**
      * The shutdown task for the fulmine runtime.
      * 
      * @author Ramon Servadei
      * 
      */
     private final class ShutdownTask implements Runnable
     {
         public void run()
         {
             FulmineContext.this.destroy();
         }
     }
 
     /** The permission profile for the context. */
     private DefaultPermissionProfile permissionProfile;
 
     /** The network */
     INetwork network;
 
     /** The connection broker */
     IConnectionBroker connectionBroker;
 
     /** The discoverer for connections */
     IConnectionDiscoverer discoverer;
 
     /** Collaborating component for model management */
     private final IModelManager modelManager;
 
     /** Collaborating component for event management */
     private final IEventManager eventManager;
 
     /** Collaborating component for distribution management */
     private IDistributionManager distributionManager;
 
     /** The internal RPC manager */
     private final IRpcManager rpcManager;
 
     /** The identity */
     private String identity;
 
     /**
      * The hashcode for this context.
      * 
      * @see FulmineContext#getContextHashCode()
      */
     private final int hashCode;
 
     /** The shutdown thread */
     private final Thread shutdownThread;
 
     /** The component that can report on the state of this context */
     private IContextWatchdog watchdog;
 
     /** The event processor identity prefix */
     private final String eventProcessorIdentityPrefix;
 
     /**
      * The remote update handler (not great that it is a concrete reference
      * rather than an interface)
      */
     private DelegatingRemoteUpdateHandler handler;
 
     /** The system info record for this context */
     private IContainer systemInfo;
 
     /**
      * Construct a context with an identity.
      * <p>
      * This version checks the system runtime for the number of processors and
      * will use that number + 1 when constructing the {@link EventManager}.
      * 
      * @param identity
      *            the context's identity
      */
     public FulmineContext(String identity)
     {
         this(identity, identity);
     }
 
     /**
      * Construct a context with an identity and the number of processors.
      * 
      * @param identity
      *            the context's identity
      * @param processors
      *            the number of processors available for events
      */
     public FulmineContext(String identity, int processors)
     {
         this(identity, identity, processors);
     }
 
     /**
      * Construct a context with an identity and the prefix for the name for the
      * event processors. If the prefix is EMPTY_STRING or the same as the
      * identity of the context, the event processor identity will simply be the
      * context identity, otherwise the event processor identity is the prefix
      * plus the context identity.
      * <p>
      * This version checks the system runtime for the number of processors and
      * will use that number + 1 when constructing the {@link EventManager}.
      * 
      * @param identity
      *            the context's identity
      * @param eventProcessorIdentityPrefix
      *            the prefix for the name for the event processors
      */
     public FulmineContext(String identity, String eventProcessorIdentityPrefix)
     {
         this(identity, eventProcessorIdentityPrefix,
             java.lang.Runtime.getRuntime().availableProcessors() + 1);
     }
 
     /**
      * Construct a context with an identity, the prefix for the name for the
      * event processors and the number of processors.
      * 
      * @param identity
      *            the context's identity
      * @param eventProcessorIdentityPrefix
      *            the prefix for the name for the event processors
      * @param processors
      *            the number of processors available for events
      */
     public FulmineContext(String identity, String eventProcessorIdentityPrefix,
         int processors)
     {
         super();
         this.permissionProfile = new DefaultPermissionProfile();
         this.shutdownThread =
             new Thread(new ShutdownTask(), identity + " context shutdown task");
         java.lang.Runtime.getRuntime().addShutdownHook(this.shutdownThread);
         nullCheck(identity, "Null identity");
         nullCheck(eventProcessorIdentityPrefix,
             "Null eventProcessorIdentityPrefix");
         this.identity = identity;
         // determine the event processor identity
         if (identity.trim().equals(eventProcessorIdentityPrefix.trim()))
         {
             this.eventProcessorIdentityPrefix = identity;
         }
         else
         {
             if (eventProcessorIdentityPrefix.trim().endsWith(COLON))
             {
                 this.eventProcessorIdentityPrefix =
                     eventProcessorIdentityPrefix.trim() + identity;
             }
             else
             {
                 this.eventProcessorIdentityPrefix =
                     eventProcessorIdentityPrefix.trim() + COLON + identity;
             }
         }
         this.eventManager =
             new EventManager(this.eventProcessorIdentityPrefix,
                 (byte) processors, this);
         this.distributionManager = new DistributionManager(this);
         this.modelManager = new ModelManager(this);
 
         try
         {
             int netName =
                 (InetAddress.getLocalHost().getHostAddress()).hashCode();
             this.hashCode =
                 new HashCodeBuilder().append(identity).append(
                     System.identityHashCode(this)).append(netName).append(
                     System.nanoTime()).toHashCode();
         }
         catch (UnknownHostException e)
         {
             throw new RuntimeException("Could not get localhost for " + this);
         }
         this.rpcManager = new RpcManager(this);
     }
 
     public IContextWatchdog getContextWatchdog()
     {
         return this.watchdog;
     }
 
     public void setContextWatchdog(IContextWatchdog watchdog)
     {
         this.watchdog = watchdog;
     }
 
     protected IModelManager getModelManager()
     {
         return this.modelManager;
     }
 
     protected IEventManager getEventManager()
     {
         return this.eventManager;
     }
 
     /**
      * Provide a different implementation of the distribution manager.
      * 
      * @param distributionManager
      *            the distribution manager to use
      */
     public void setDistributionManager(IDistributionManager distributionManager)
     {
         // destroy the previous manager
         if (this.distributionManager != null)
         {
             getDistributionManager().destroy();
         }
         this.distributionManager = distributionManager;
     }
 
     public IDistributionManager getDistributionManager()
     {
         return this.distributionManager;
     }
 
     public IPermissionProfile getPermissionProfile()
     {
         return this.permissionProfile;
     }
 
     public void setPermissionProfile(IPermissionProfile permissionProfile)
     {
         this.permissionProfile.setApplicationPermissions(permissionProfile);
     }
 
     @Override
     protected void doStart()
     {
         getEventManager().start();
         getModelManager().start();
         getRpcManager().start();
         getDistributionManager().start();
         // create the system info record
         final IContainer systemInfo =
             getLocalContainer(getIdentity() + ":SystemInfo", Type.SYSTEM,
                 Domain.FRAMEWORK);
         this.systemInfo = systemInfo;
         this.systemInfo.add(new LongField(SystemInfoFields.FREE_MEMORY));
         this.systemInfo.add(new LongField(SystemInfoFields.TOTAL_MEMORY));
         // schedule a task to flush the system info record periodically
         schedule(new TimerTask()
         {
             @Override
             public void run()
             {
                 EventQueueItem.isStatisticsEvent.set(Boolean.TRUE);
                 try
                 {
                     FulmineContext.this.systemInfo.beginFrame(new EventFrameExecution());
                     try
                     {
                         FulmineContext.this.systemInfo.getLongField(
                             SystemInfoFields.FREE_MEMORY).set(
                             Runtime.getRuntime().freeMemory() / 1000);
                         FulmineContext.this.systemInfo.getLongField(
                             SystemInfoFields.TOTAL_MEMORY).set(
                             Runtime.getRuntime().totalMemory() / 1000);
                     }
                     finally
                     {
                         FulmineContext.this.systemInfo.flushFrame();
                     }
                 }
                 finally
                 {
                     EventQueueItem.isStatisticsEvent.remove();
                 }
             }
         }, 10000, 10000);
         if (getNetwork() != null)
         {
             setConnectionBroker(getNetwork().createBroker());
             setConnectionDiscoverer(getNetwork().createDiscoverer());
             getConnectionBroker().start();
             if (getNetwork().isListeningOnlyMode())
             {
                 getConnectionDiscoverer().disablePulsing();
             }
             getConnectionDiscoverer().start();
 
             if (getHandler().getDelegate() == null)
             {
                 getHandler().setDelegate(new VetoingRemoteUpdateHandler());
             }
             // publish the remote update handler RPC method
             getRpcManager().publishRpcs(IRemoteUpdateHandler.class,
                 this.getHandler());
         }
         if (getLog().isInfoEnabled())
         {
             getLog().info("Started context " + this);
         }
     }
 
     @Override
     protected void doDestroy()
     {
         /*
          * The discoverer and broker may have implementations that do not extend
          * the AbstractLifeCycle (which guarantees no exception from destroy())
          * so we call the destroy methods of these in individual try-catch
          * blocks.
          */
         if (getConnectionDiscoverer() != null)
         {
             try
             {
                 getConnectionDiscoverer().destroy();
             }
             catch (Exception e)
             {
                 logException(getLog(), "discoverer", e);
             }
         }
         if (getConnectionBroker() != null)
         {
             try
             {
                 getConnectionBroker().destroy();
             }
             catch (Exception e)
             {
                 logException(getLog(), "connectionBroker", e);
             }
         }
         if (getNetwork() != null)
         {
             getNetwork().setContext(null);
         }
         getDistributionManager().destroy();
         getRpcManager().destroy();
         getModelManager().destroy();
         getEventManager().destroy();
         if (Thread.currentThread() != this.shutdownThread)
         {
             Runtime.getRuntime().removeShutdownHook(this.shutdownThread);
         }
     }
 
     public void addContainer(IContainer container)
     {
         getModelManager().addContainer(container);
     }
 
     public String getIdentity()
     {
         return this.identity;
     }
 
     public void setNetwork(INetwork network)
     {
         if (isActive())
         {
             throw new IllegalStateException(
                 "Must be called before starting the context");
         }
         this.network = network;
         this.network.setContext(this);
     }
 
     public INetwork getNetwork()
     {
         return this.network;
     }
 
     public IContainer getLocalContainer(String identity, IType type,
         IDomain domain)
     {
         return getModelManager().getLocalContainer(identity, type, domain);
     }
 
     public IContainer getRemoteContainer(String remoteContextIdentity,
         String containerIdentity, IType type, IDomain domain)
     {
         return getModelManager().getRemoteContainer(remoteContextIdentity,
             containerIdentity, type, domain);
     }
 
     public boolean containsRemoteContainer(String remoteContextIdentity,
         String containerIdentity, IType type, IDomain domain)
     {
         return getModelManager().containsRemoteContainer(remoteContextIdentity,
             containerIdentity, type, domain);
     }
 
     public boolean containsLocalContainer(String containerIdentity, IType type,
         IDomain domain)
     {
         return getModelManager().containsLocalContainer(containerIdentity,
             type, domain);
     }
 
     public void queueEvent(IEvent event)
     {
         getEventManager().queueEvent(event);
     }
 
     public void queueEvents(Collection<IEvent> events)
     {
         getEventManager().queueEvents(events);
     }
 
     public boolean removeContainer(IContainer container)
     {
         return getModelManager().removeContainer(container);
     }
 
     public boolean subscribe(String contextIdentity, String identityPattern,
         IType type, IDomain domain, IEventListener listener)
     {
         return getDistributionManager().subscribe(contextIdentity,
             identityPattern, type, domain, listener);
     }
 
     public boolean unsubscribe(String contextIdentity, String identityPattern,
         IType type, IDomain domain, IEventListener listener)
     {
         return getDistributionManager().unsubscribe(contextIdentity,
             identityPattern, type, domain, listener);
     }
 
     public IChannel[] getConnectedChannels()
     {
         return getDistributionManager().getConnectedChannels();
     }
 
     public int getEventProcessorCount()
     {
         return getEventManager().getEventProcessorCount();
     }
 
     public Collection<IContainer> getLocalContainers()
     {
         return getModelManager().getLocalContainers();
     }
 
     public Collection<IContainer> getRemoteContainers(
         String remoteContextIdentity)
     {
         return getModelManager().getRemoteContainers(remoteContextIdentity);
     }
 
     public ThreadGroup getEventProcessorThreadGroup()
     {
         return getEventManager().getEventProcessorThreadGroup();
     }
 
     public IConnectionBroker getConnectionBroker()
     {
         return this.connectionBroker;
     }
 
     public void setConnectionBroker(IConnectionBroker broker)
     {
         this.connectionBroker = broker;
     }
 
     public IConnectionDiscoverer getConnectionDiscoverer()
     {
         return this.discoverer;
     }
 
     public void setConnectionDiscoverer(IConnectionDiscoverer discoverer)
     {
         this.discoverer = discoverer;
     }
 
     public IContainerFactory getContainerFactory()
     {
         return getModelManager().getContainerFactory();
     }
 
     public ISystemEventSource getSystemEventSource(
         Class<? extends ISystemEvent> type)
     {
         return getEventManager().getSystemEventSource(type);
     }
 
     public void execute(Runnable task)
     {
         getEventManager().execute(task);
     }
 
     public IFrameReader getFrameReader()
     {
         return getDistributionManager().getFrameReader();
     }
 
     public IFrameWriter getFrameWriter()
     {
         return getDistributionManager().getFrameWriter();
     }
 
     public void requestRetransmit(String contextIdentity,
         String identityRegularExpression, IType type, IDomain domain)
     {
         getDistributionManager().requestRetransmit(contextIdentity,
             identityRegularExpression, type, domain);
     }
 
     public void requestRetransmitAll(String contextIdentity)
     {
         getDistributionManager().requestRetransmitAll(contextIdentity);
     }
 
     public void retransmit(String contextIdentity,
         String identityRegularExpression, IType type, IDomain domain)
     {
         getDistributionManager().retransmit(contextIdentity,
             identityRegularExpression, type, domain);
     }
 
     public void retransmitAll(String contextIdentity)
     {
         getDistributionManager().retransmitAll(contextIdentity);
     }
 
     public void retransmitAllToAll()
     {
         getDistributionManager().retransmitAllToAll();
     }
 
     public void retransmitToAll(String identityRegularExpression, IType type,
         IDomain domain)
     {
         getDistributionManager().retransmitToAll(identityRegularExpression,
             type, domain);
     }
 
     public boolean addSubscriptionListener(ISubscriptionListener listener)
     {
         return getDistributionManager().addSubscriptionListener(listener);
     }
 
     public boolean removeSubscriptionListener(ISubscriptionListener listener)
     {
         return getDistributionManager().removeSubscriptionListener(listener);
     }
 
     public int getContextHashCode()
     {
         return this.hashCode;
     }
 
     public String getEventProcessorIdentityPrefix()
     {
         return this.eventProcessorIdentityPrefix;
     }
 
     public void invokeRpc(String remoteContextIdentity, byte[] rpcData)
     {
         getDistributionManager().invokeRpc(remoteContextIdentity, rpcData);
     }
 
     public boolean addRpcPublicationListener(String remoteContextIdentity,
         IRpcPublicationListener listener)
     {
         return getRpcManager().addRpcPublicationListener(remoteContextIdentity,
             listener);
     }
 
     public IRpcMarker invoke(IRpcResultHandler resultHandler,
         String remoteContextIdentity, String procedure, IField... args)
     {
         return getRpcManager().invoke(resultHandler, remoteContextIdentity,
             procedure, args);
     }
 
     public IRpcResult invoke(String remoteContextIdentity, String procedure,
         IField... args)
     {
         return getRpcManager().invoke(remoteContextIdentity, procedure, args);
     }
 
     public boolean publishProdedure(IRpcHandler handler,
         IRpcDefinition rpcDefinition)
     {
         return getRpcManager().publishProdedure(handler, rpcDefinition);
     }
 
     public boolean unpublishProdedure(IRpcDefinition rpcDefinition)
     {
         return getRpcManager().unpublishProdedure(rpcDefinition);
     }
 
     public boolean unpublishRpcs(Class<?> definition, Object handler)
     {
         return getRpcManager().unpublishRpcs(definition, handler);
     }
 
     public boolean publishRpcs(Class<?> definition, Object handler)
     {
         return getRpcManager().publishRpcs(definition, handler);
     }
 
     public boolean removeRpcPublicationListener(String remoteContextIdentity,
         IRpcPublicationListener listener)
     {
         return getRpcManager().removeRpcPublicationListener(
             remoteContextIdentity, listener);
     }
 
     public void setRemoteUpdateHandler(IRemoteUpdateHandler handler)
     {
         handler.setContext(this);
         getHandler().setDelegate(handler);
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     private IRpcManager getRpcManager()
     {
         return this.rpcManager;
     }
 
     @Override
     public final String toString()
     {
         return super.toString() + COLON + getIdentity() + OPEN_BRACE
             + hashCode() + CLOSE_BRACE;
     }
 
     @Override
     public final boolean equals(Object obj)
     {
         return super.equals(obj);
     }
 
     @Override
     public final int hashCode()
     {
         return this.hashCode;
     }
 
     public String updateRemoteContainer(String remoteContextIdentity,
         String identity, IType type, IDomain domain, String fieldName,
         String fieldValueAsString)
     {
         return getDistributionManager().updateRemoteContainer(
             remoteContextIdentity, identity, type, domain, fieldName,
             fieldValueAsString);
     }
 
     /**
      * Lazily constructs the {@link DelegatingRemoteUpdateHandler}
      * 
      * @return the {@link DelegatingRemoteUpdateHandler}
      */
     private DelegatingRemoteUpdateHandler getHandler()
     {
         if (this.handler == null)
         {
             this.handler = new DelegatingRemoteUpdateHandler();
         }
         return handler;
     }
 
     public IContainer getSystemInfo()
     {
         return this.systemInfo;
     }
 
     public void schedule(TimerTask task, long delay, long period)
     {
         getEventManager().schedule(task, delay, period);
     }
 }