/*
    * 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 net.sourceforge.basher.internal.impl;
  
  import java.util.HashSet;
  import java.util.Set;
  import java.util.concurrent.Semaphore;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  import net.sourceforge.basher.Basher;
  import net.sourceforge.basher.TaskExecutionContext;
  import net.sourceforge.basher.TaskManager;
  import net.sourceforge.basher.events.*;
  import net.sourceforge.basher.internal.Randomizer;
  import net.sourceforge.basher.internal.TaskInvoker;
  import net.sourceforge.basher.internal.TaskRunner;
  import org.apache.commons.logging.Log;
  import org.ops4j.gaderian.events.RegistryShutdownListener;
  
  /**
   * @author Johan Lindquist
   * @version 1.0
   */
  public class TaskRunnerImpl implements TaskRunner, BasherEventListener, RegistryShutdownListener
  {
      private Randomizer _randomizer;
      private TaskManager _taskManager;
      private TaskInvoker _taskInvoker;
      private Log _logger;
      private int _minTime;
      private int _maxDelay;
  
      private final Set<String> _threadsShutdown = new HashSet<String>();
  
      private final Semaphore _semaphore = new Semaphore(0);
  
      private final AtomicBoolean _keepRunning = new AtomicBoolean(true);
      public int numberOfThreadsRunning = 0;
  
      public void setRandomizer(final Randomizer randomizer)
      {
          _randomizer = randomizer;
      }
  
      public void setLog(final Log logger)
      {
          _logger = logger;
      }
  
      public void setTaskManager(final TaskManager taskManager)
      {
          _taskManager = taskManager;
      }
  
      public void setTaskInvoker(final TaskInvoker taskInvoker)
      {
          _taskInvoker = taskInvoker;
      }
  
  
      public void stopInvoking()
      {
          // And cancel running too
          _keepRunning.set(false);
      }
  
      /**
       *
       */
      public void run()
      {
          // This call actually blocks if we are not yet running
          while (keepProcessing())
          {
              try
              {
                  // Retrieve the next task to run
                  final TaskExecutionContext taskExecutionContext = _taskManager.getNextTaskExecutionContext();
  
                  // Task Manager API will return null if not tasks are available
                  if ( taskExecutionContext != null)
                  {
                      _taskInvoker.invokeTask( taskExecutionContext );
                  }
  
                  // Simple check to ensure we dont overdo things ...
                  if (_keepRunning.get())
                 {
                     Basher.fireCleanUpThread();
                     Thread.sleep(getRandomSleepTime());
                 }
             }
             catch (final Throwable e)
             {
                 _logger.error(e.getMessage(), e);
             }
         }
     }
 
     private boolean keepProcessing()
     {
         try
         {
             _semaphore.acquire();
             try
             {
                 // Once we wake up, check if we need to shutdown
                 final String threadName = Thread.currentThread().getName();
                 if (_threadsShutdown.contains(threadName))
                 {
                     _threadsShutdown.remove(threadName);
                     // Remove one more permit
                     _semaphore.acquire();
                     return false;
                 }
                 return _keepRunning.get();
             }
             finally
             {
                 _semaphore.release();
             }
 
         }
         catch (final InterruptedException e)
         {
             //
         }
         return false;
     }
 
     private long getRandomSleepTime()
     {
         // Should create (_minTime < sleepTime < (_minTime + _maxDelay))
         return _minTime + _randomizer.getRandomInt(_maxDelay);
     }
 
     public void basherEvent(final BasherEvent basherEvent)
     {
         if (basherEvent instanceof PhaseTransitionEvent)
         {
             // Anything?
             final PhaseTransitionEvent phaseTransitionEvent = (PhaseTransitionEvent) basherEvent;
             _maxDelay = phaseTransitionEvent.getBasherContext().getTaskMaxDelay();
             _minTime = phaseTransitionEvent.getBasherContext().getTaskMinDelay();
 
             switch (phaseTransitionEvent.getNewPhase())
             {
                 case START:
                     numberOfThreadsRunning = phaseTransitionEvent.getBasherContext().getInitialNumberThreads();
                     break;
                 case SETUP:
                     break;
                 case RUN:
                     // Release all waiting threads
                     _logger.info("Number of available permits: " + _semaphore.availablePermits());
                     _logger.info("Number of threads: " + numberOfThreadsRunning);
                     _semaphore.release(numberOfThreadsRunning);
                     break;
                 case COOLDOWN:
                 case TEARDOWN:
                     try
                     {
                         _semaphore.acquire(_semaphore.availablePermits());
                     }
                     catch (final InterruptedException e)
                     {
                         e.printStackTrace();
                     }
                     break;
                 case END:
                     flipProcessing(false);
                 default:
             }
         }
         else if (basherEvent instanceof ThreadAddedEvent)
         {
             numberOfThreadsRunning = ((ThreadAddedEvent)basherEvent).getCurrentNumberThreads();
             _semaphore.release(1);
         }
         else if (basherEvent instanceof ThreadRemovedEvent)
         {
             final ThreadRemovedEvent threadRemovedEvent = (ThreadRemovedEvent) basherEvent;
             numberOfThreadsRunning = threadRemovedEvent.getCurrentNumberThreads();
             final String threadName = threadRemovedEvent.getName();
             // Simple check to see if the thread is already awaiting shutdown
             if (!_threadsShutdown.contains(threadName))
             {
                 _logger.info("Thread " + threadName + " shutting down on request");
                 _threadsShutdown.add(threadName);
             }
         }
         else if (basherEvent instanceof NoTasksAvailableEvent)
         {
             flipProcessing(false);
         }
         else if (basherEvent instanceof TasksAvailableEvent)
         {
             // Release all the waiting threads
             flipProcessing(true);
         }
     }
 
     private void flipProcessing(final boolean state)
     {
         if (state)
         {
             _semaphore.release(numberOfThreadsRunning);
         }
         else
         {
             if (_semaphore.availablePermits() == 0)
             {
                 // No need to acquire any - all threads are holding at this stage
             }
             else
             {
                 try
                 {
                     _semaphore.acquire(_semaphore.availablePermits());
                 }
                 catch (final InterruptedException e)
                 {
                     e.printStackTrace();
                 }
             }
         }
     }
 
     public void registryDidShutdown()
     {
         stopInvoking();
     }
 }