/*
      Copyright 2008 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.COLON;
  import static fulmine.util.Utils.SPACING_4_CHARS;
  import static fulmine.util.Utils.logException;
  
  import java.util.Collection;
  import java.util.TimerTask;
  
  import org.apache.commons.lang.SystemUtils;
  
  import fulmine.AbstractLifeCycle;
  import fulmine.event.EventProcessor;
  import fulmine.event.IEvent;
  import fulmine.event.IEventManager;
  import fulmine.event.IEventSource;
  import fulmine.event.system.ISystemEvent;
  import fulmine.event.system.ISystemEventSource;
  import fulmine.event.system.SystemEventSource;
  import fulmine.model.container.IContainer;
  import fulmine.model.field.IField;
  import fulmine.util.concurrent.ITaskExecutor;
  import fulmine.util.concurrent.TaskExecutor;
  import fulmine.util.log.AsyncLog;
  import fulmine.util.reference.AutoCreatingStore;
  import fulmine.util.reference.IAutoCreatingStore;
  import fulmine.util.reference.IObjectBuilder;
  
  /**
   * A manager for the event framework of a context. This constructs a number of
   * internal {@link EventProcessor} instances that handle {@link IEvent}
   * distribution.
   * <p>
   * The allocation of an event to an event processor is based on the
   * {@link IEventSource#getEventSourceGroupId} of the event's
   * {@link IEventSource} <code>modulo</code> the number of processors. This
   * ensures events from the same source are processed by the same processor and
   * thus listeners registered with a source are guaranteed to receive events
   * originating from an event source in the order they occurred.
   * <p>
   * This class also has a single threaded task executor that is used to service
   * requests via the {@link #execute(Runnable)} method. This executor is meant to
   * be used as a utility for executing "one-shot" asynchronous tasks.
   * 
   * @see IFulmineContext
   * @author Ramon Servadei
   */
  final class EventManager extends AbstractLifeCycle implements IEventManager
  {
      private final static AsyncLog LOG = new AsyncLog(EventManager.class);
  
      /** The basename for an event processor */
      static final String PROCESSOR_BASENAME =
          IFrameworkContext.NAME + "EventProcessor";
  
      /**
       * Builder for {@link SystemEventSource} instances
       * 
       * @author Ramon Servadei
       * 
       */
      @SuppressWarnings("unchecked")
      private static final class SystemEventSourceBuilder implements
          IObjectBuilder<Class, ISystemEventSource>
      {
          public ISystemEventSource create(Class key)
          {
              final SystemEventSource systemEventSource =
                  new SystemEventSource(key.getSimpleName());
              systemEventSource.start();
              return systemEventSource;
          }
      }
  
      /** The event processor thread group */
      final ThreadGroup eventProcessorThreadGroup;
  
      /** The {@link EventProcessor} instances in the context */
      protected final EventProcessor[] processors;
  
      /** Executor for 'one-shot' tasks */
      private final ITaskExecutor executor;
  
      private final byte availableProcessors;
 
     /** The typed event sources */
     @SuppressWarnings("unchecked")
     private final IAutoCreatingStore<Class, ISystemEventSource> eventSources;
 
     /**
      * Standard constructor
      * 
      * @param identity
      *            the identity for the threads
      * @param eventProcessorCount
      *            the number of processors available
      * @param context
      *            required to construct the information container records for
      *            the {@link EventProcessor} instances. The containers are
      *            created lazily.
      */
     @SuppressWarnings("unchecked")
     EventManager(final String identity, byte eventProcessorCount,
         IFulmineContext context)
     {
         super();
         if (eventProcessorCount < 2)
         {
             throw new IllegalArgumentException(
                 "eventProcessorCount cannot be < 2");
         }
         this.availableProcessors = eventProcessorCount;
 
         this.eventProcessorThreadGroup =
             createThreadGroup(identity, Thread.NORM_PRIORITY);
 
         this.processors = new EventProcessor[this.availableProcessors];
         createProcessors(this.processors,
             identity + COLON + PROCESSOR_BASENAME,
             this.eventProcessorThreadGroup, context);
 
         this.executor =
             new TaskExecutor(identity + COLON + "AdhocProcessor", context);
         this.eventSources =
             new AutoCreatingStore<Class, ISystemEventSource>(
                 new SystemEventSourceBuilder());
         if (getLog().isInfoEnabled())
         {
             getLog().info(
                 "Created " + this.availableProcessors + " event processors");
         }
     }
 
     /**
      * Utility method to create a thread group
      * 
      * @param identity
      *            the identity for the thread group
      * @param priority
      *            the priority of the thread group
      * @return the {@link ThreadGroup}
      */
     private ThreadGroup createThreadGroup(String identity, int priority)
     {
         String ps;
         switch (priority)
         {
             case Thread.MIN_PRIORITY:
                 ps = "LowPriority";
                 break;
             case Thread.MAX_PRIORITY:
                 ps = "HighPriority";
                 break;
             case Thread.NORM_PRIORITY:
                 ps = "NormalPriority";
                 break;
             default:
                 throw new RuntimeException("Unknown priority: " + priority);
         }
         final ThreadGroup threadGroup =
             new ThreadGroup(identity + COLON + IFrameworkContext.NAME + ps);
         threadGroup.setMaxPriority(priority);
         return threadGroup;
     }
 
     public ThreadGroup getEventProcessorThreadGroup()
     {
         return this.eventProcessorThreadGroup;
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     @Override
     protected void doStart()
     {
         this.executor.start();
         startProcessors(this.processors);
         if (getLog().isInfoEnabled())
         {
             getLog().info("Started");
         }
     }
 
     @Override
     protected void doDestroy()
     {
         this.eventSources.destroy();
         // destroy the processors
         this.executor.destroy();
         destroyProcessors(this.processors);
         // destroy the thread groups
         destroyThreadGroup(this.eventProcessorThreadGroup);
     }
 
     public void queueEvents(Collection<IEvent> events)
     {
         if (events != null)
         {
             for (IEvent event : events)
             {
                 try
                 {
                     queueEvent(event);
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), event, e);
                 }
             }
         }
     }
 
     public void queueEvent(IEvent event)
     {
         checkActive();
         if (event != null)
         {
             // find the processor for the event SOURCE
             final IEventSource source = event.getSource();
             if (source != null)
             {
                 int eventProcessorId = 0;
                 if (source instanceof IField)
                 {
                     // ensures fields use the same processor as their container
                     final IContainer container =
                         ((IField) source).getContainer();
                     if (container == null)
                     {
                         return;
                     }
                     eventProcessorId = container.getEventSourceGroupId();
                 }
                 else
                 {
                     eventProcessorId = source.getEventSourceGroupId();
                 }
                 final EventProcessor processor =
                     this.processors[eventProcessorId];
                 if (processor.isActive())
                 {
                     if (getLog().isTraceEnabled())
                     {
                         getLog().trace(
                             "Adding " + SystemUtils.LINE_SEPARATOR
                                 + SPACING_4_CHARS + "Event: "
                                 + event.toString() + SystemUtils.LINE_SEPARATOR
                                 + SPACING_4_CHARS + "Source: "
                                 + event.getSource().toIdentityString()
                                 + SystemUtils.LINE_SEPARATOR + SPACING_4_CHARS
                                 + "Processor: " + processor, new Exception());
                     }
                     else
                     {
                         if (getLog().isDebugEnabled())
                         {
                             getLog().debug(
                                 "Adding " + SystemUtils.LINE_SEPARATOR
                                     + SPACING_4_CHARS + "Event: "
                                     + event.toIdentityString()
                                     + SystemUtils.LINE_SEPARATOR
                                     + SPACING_4_CHARS + "Source: "
                                     + event.getSource().toIdentityString()
                                     + SystemUtils.LINE_SEPARATOR
                                     + SPACING_4_CHARS + "Processor: "
                                     + processor);
                         }
                     }
                     processor.queue(event);
                 }
             }
         }
     }
 
     public int getEventProcessorCount()
     {
         return this.availableProcessors;
     }
 
     private void createProcessors(EventProcessor[] processors,
         String processorName, ThreadGroup threadGroup, IFulmineContext context)
     {
         for (int i = 0; i < processors.length; i++)
         {
             processors[i] =
                 new EventProcessor(processorName + i, threadGroup, context);
         }
     }
 
     private void destroyThreadGroup(ThreadGroup threadGroup)
     {
         // interrupt all threads we know about
         Thread[] threads = new Thread[threadGroup.activeCount()];
         threadGroup.enumerate(threads);
         for (Thread thread : threads)
         {
             try
             {
                 if (!Thread.currentThread().equals(thread) && thread != null)
                 {
                     thread.interrupt();
                 }
             }
             catch (Exception e)
             {
                 logException(getLog(), thread, e);
             }
         }
     }
 
     private void destroyProcessors(EventProcessor[] eventProcessors)
     {
         for (EventProcessor processor : eventProcessors)
         {
             try
             {
                 processor.destroy();
             }
             catch (Exception e)
             {
                 logException(getLog(), processor, e);
             }
         }
     }
 
     private void startProcessors(EventProcessor[] eventProcessors)
     {
         for (EventProcessor processor : eventProcessors)
         {
             processor.start();
         }
     }
 
     public ISystemEventSource getSystemEventSource(
         Class<? extends ISystemEvent> type)
     {
         return this.eventSources.get(type);
     }
 
     public void execute(Runnable task)
     {
         checkActive();
         this.executor.execute(task);
     }
 
     public void schedule(TimerTask task, long delay, long period)
     {
         checkActive();
         this.executor.schedule(task, delay, period);
     }
 }