/*
    * Copyright (c) 2002-2022, City of Paris
    * All rights reserved.
    *
    * Redistribution and use in source and binary forms, with or without
    * modification, are permitted provided that the following conditions
    * are met:
    *
    *  1. Redistributions of source code must retain the above copyright notice
   *     and the following disclaimer.
   *
   *  2. Redistributions in binary form must reproduce the above copyright notice
   *     and the following disclaimer in the documentation and/or other materials
   *     provided with the distribution.
   *
   *  3. Neither the name of 'Mairie de Paris' nor 'Lutece' nor the names of its
   *     contributors may be used to endorse or promote products derived from
   *     this software without specific prior written permission.
   *
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
   * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
   * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE
   * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
   * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
   * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
   * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
   * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   * POSSIBILITY OF SUCH DAMAGE.
   *
   * License 1.0
   */
  package fr.paris.lutece.portal.service.daemon;
  
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Set;
  import java.util.Timer;
  import java.util.TimerTask;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.TimeUnit;
  
  import fr.paris.lutece.portal.service.util.AppLogService;
  import fr.paris.lutece.portal.service.util.AppPropertiesService;
  
  /**
   * Daemon scheduler.
   * <p>
   * Responsible for ensuring on demand or timely daemon execution. Starts a thread which monitor the queue for daemons to execute. A {@link Timer} handles
   * repeating daemons runs.
   * 
   * <p>
   * Daemon run requests are coalesced. If a daemon is already running when a request comes, a new run is scheduled right after the current run ends.
   */
  class DaemonScheduler implements Runnable, IDaemonScheduler
  {
      private static final String PROPERTY_MAX_AWAIT_TERMINATION_DELAY = "daemon.maxAwaitTerminationDelay";
  
      private final BlockingQueue<DaemonEntry> _queue;
      private final ExecutorService _executor;
      private final Thread _coordinatorThread;
      private final Timer _scheduledDaemonsTimer;
      private final Map<String, RunnableWrapper> _executingDaemons;
      private final Map<String, DaemonTimerTask> _scheduledDaemons;
      private volatile boolean _bShuttingDown;
  
      /**
       * Constructor
       * 
       * @param queue
       *            the queue where daemon execution requests are stored
       * @param executor
       *            the executor service handling the execution of daemons
       */
      public DaemonScheduler( BlockingQueue<DaemonEntry> queue, ExecutorService executor )
      {
          _queue = queue;
          _executor = executor;
          _scheduledDaemonsTimer = new Timer( "Lutece-Daemons-Scheduled-Timer-Thread", true );
          _executingDaemons = new HashMap<>( );
          _scheduledDaemons = new HashMap<>( );
          _coordinatorThread = new Thread( this, "Lutece-Daemons-Coordinator" );
          _coordinatorThread.setDaemon( true );
          _coordinatorThread.start( );
          _bShuttingDown = false;
      }
  
      @Override
      public boolean enqueue( DaemonEntry entry, long nDelay, TimeUnit unit )
      {
          assertNotShuttingDown( );
          if ( nDelay == 0L )
          {
              boolean queued = _queue.offer( entry );
              if ( !queued )
             {
                 AppLogService.error( "Failed to enqueue a run of daemon {}", entry.getId( ) );
             }
             return queued;
         }
         try
         {
             _scheduledDaemonsTimer.schedule( new DaemonTimerTask( entry ), unit.toMillis( nDelay ) );
             return true;
         }
         catch( IllegalStateException e )
         {
             return false;
         }
     }
 
     private void assertNotShuttingDown( )
     {
         if ( _bShuttingDown )
         {
             throw new IllegalStateException( "DaemonScheduler is shutting down. Enqueing tasks or scheduling tasks is not possible anymore." );
         }
     }
 
     /**
      * Enqueue without delay
      * 
      * @param entry
      *            the daemon entry
      */
     private void enqueue( DaemonEntry entry )
     {
         enqueue( entry, 0L, TimeUnit.MILLISECONDS );
     }
 
     @Override
     public void schedule( DaemonEntry entry, long nInitialDelay, TimeUnit unit )
     {
         assertNotShuttingDown( );
         synchronized( _scheduledDaemons )
         {
             if ( _scheduledDaemons.containsKey( entry.getId( ) ) )
             {
                 AppLogService.error( "Daemon " + entry.getId( ) + " already scheduled, not scheduling again" );
             }
             else
             {
 
                 DaemonTimerTask daemonTimerTask = new DaemonTimerTask( entry );
                 _scheduledDaemonsTimer.scheduleAtFixedRate( daemonTimerTask, unit.toMillis( nInitialDelay ), entry.getInterval( ) * 1000 );
                 _scheduledDaemons.put( entry.getId( ), daemonTimerTask );
             }
         }
 
     }
 
     @Override
     public void unSchedule( DaemonEntry entry )
     {
         synchronized( _scheduledDaemons )
         {
             DaemonTimerTask daemonTimerTask = _scheduledDaemons.get( entry.getId( ) );
             if ( daemonTimerTask == null )
             {
                 AppLogService.error( "Could not unschedule daemon " + entry.getId( ) + " which was not scheduled" );
             }
             else
             {
                 daemonTimerTask.cancel( );
                 _scheduledDaemonsTimer.purge( );
                 _scheduledDaemons.remove( entry.getId( ) );
             }
         }
         boolean bStopScheduled = false;
         synchronized( _executingDaemons )
         {
             RunnableWrapper runnable = _executingDaemons.get( entry.getId( ) );
             if ( runnable != null )
             {
                 runnable.stopDaemonAfterExecution( );
                 bStopScheduled = true;
             }
         }
         if ( !bStopScheduled )
         {
             try
             {
                 entry.getDaemon( ).stop( );
             }
             catch( Throwable t )
             {
                 AppLogService.error( "Failed to stop daemon {}", entry.getId( ), t );
             }
         }
     }
 
     @Override
     public void run( )
     {
         // use a set to coalesce daemon signaling
         Set<DaemonEntry> queued = new HashSet<>( );
         do
         {
             try
             {
                 // collect signaled daemons
                 queued.add( _queue.take( ) );
                 _queue.drainTo( queued );
             }
             catch( InterruptedException e )
             {
                 // We were asked to stop
                 break;
             }
             // execute them
             for ( DaemonEntry entry : queued )
             {
                 RunnableWrapper runnable = null;
                 synchronized( _executingDaemons )
                 {
                     runnable = _executingDaemons.get( entry.getId( ) );
                     if ( runnable != null )
                     {
                         // already executing; schedule a new run after this one
                         runnable.shouldEnqueueAgain( );
                         runnable = null;
                     }
                     else
                     {
                         runnable = new RunnableWrapper( entry );
                         _executingDaemons.put( entry.getId( ), runnable );
                     }
                 }
                 if ( runnable != null )
                 {
                     _executor.execute( runnable );
                 }
             }
             // prepare next iteration
             queued.clear( );
         }
         while ( !Thread.interrupted( ) );
     }
 
     @Override
     public void shutdown( )
     {
         _bShuttingDown = true; // prevent future scheduling of daemons
         int maxAwaitTerminationDelay = AppPropertiesService.getPropertyInt( PROPERTY_MAX_AWAIT_TERMINATION_DELAY, 15 );
         AppLogService
                 .info( "Lutece daemons scheduler stop requested : trying to terminate gracefully daemons list (max wait " + maxAwaitTerminationDelay + " s)." );
         _scheduledDaemonsTimer.cancel( );
         _coordinatorThread.interrupt( );
         _executor.shutdown( );
 
         // make a copy of scheduled daemons so that the list can be modified by
         // #unSchedule
         ArrayList<DaemonTimerTask> scheduled = new ArrayList<>( _scheduledDaemons.values( ) );
         scheduled.forEach( task -> unSchedule( task.getDaemonEntry( ) ) );
 
         try
         {
             if ( _executor.awaitTermination( maxAwaitTerminationDelay, TimeUnit.SECONDS ) )
             {
                 AppLogService.info( "All daemons shutdown successfully." );
             }
             else
             {
                 AppLogService.info( "Some daemons are still running, trying to interrupt them..." );
                 _executor.shutdownNow( );
 
                 if ( _executor.awaitTermination( 1, TimeUnit.SECONDS ) )
                 {
                     AppLogService.info( "All running daemons successfully interrupted." );
                 }
                 else
                 {
                     AppLogService.error( "Interrupt failed; daemons still running." );
                 }
             }
         }
         catch( InterruptedException e )
         {
             AppLogService.error( "Interruped while waiting for daemons termination", e );
         }
     }
 
     /**
      * Wrapper for the daemon Runnable which can enqueue a new run when execution completes
      */
     private final class RunnableWrapper implements Runnable
     {
 
         private final DaemonEntry _entry;
         private volatile boolean _bShouldEnqueueAgain;
         private volatile boolean _bstopAfterExecution;
 
         /**
          * @param entry
          *            the wrapped DaemonEntry
          */
         public RunnableWrapper( DaemonEntry entry )
         {
             _entry = entry;
         }
 
         public void stopDaemonAfterExecution( )
         {
             _bstopAfterExecution = true;
         }
 
         /**
          * Signals that an execution of the daemon should be enqueued on completion
          */
         public void shouldEnqueueAgain( )
         {
             _bShouldEnqueueAgain = true;
         }
 
         @Override
         public void run( )
         {
             try
             {
                 _entry.getDaemonThread( ).run( );
             }
             finally
             {
                 synchronized( _executingDaemons )
                 {
                     _executingDaemons.remove( _entry.getId( ) );
                 }
                 if ( _bstopAfterExecution )
                 {
                     _entry.getDaemon( ).stop( );
                 }
                 else
                     if ( _bShouldEnqueueAgain )
                     {
                         enqueue( _entry );
                     }
             }
         }
 
     }
 
     /**
      * Timer task to enqueue daemon runs
      */
     private final class DaemonTimerTask extends TimerTask
     {
 
         private final DaemonEntry _entry;
 
         /**
          * @param entry
          *            the daemon
          */
         public DaemonTimerTask( DaemonEntry entry )
         {
             _entry = entry;
         }
 
         @Override
         public void run( )
         {
             enqueue( _entry );
         }
 
         /**
          * Access the daemon entry
          * 
          * @return the daemon
          */
         public DaemonEntry getDaemonEntry( )
         {
             return _entry;
         }
 
     }
 }