/**
    * Logback: the reliable, generic, fast and flexible logging framework.
    * Copyright (C) 1999-2015, QOS.ch. All rights reserved.
    *
    * This program and the accompanying materials are dual-licensed under
    * either the terms of the Eclipse Public License v1.0 as published by
    * the Eclipse Foundation
    *
    *   or (per the licensee's choosing)
   *
   * under the terms of the GNU Lesser General Public License version 2.1
   * as published by the Free Software Foundation.
   */
  package ch.qos.logback.core;
  
  import ch.qos.logback.core.spi.AppenderAttachable;
  import ch.qos.logback.core.spi.AppenderAttachableImpl;
  import ch.qos.logback.core.util.InterruptUtil;
  
  import java.util.ArrayList;
  import java.util.Iterator;
  import java.util.List;
  import java.util.concurrent.ArrayBlockingQueue;
  import java.util.concurrent.BlockingQueue;
  
  /**
   * This appender and derived classes, log events asynchronously. In order to
   * avoid loss of logging events, this appender should be closed. It is the
   * user's responsibility to close appenders, typically at the end of the
   * application lifecycle.
   * <p>
   * This appender buffers events in a {@link BlockingQueue}. {@link Worker}
   * thread created by this appender takes events from the head of the queue, and
   * dispatches them to the single appender attached to this appender.
   * <p>
   * Please refer to the
   * <a href="http://logback.qos.ch/manual/appenders.html#AsyncAppender">logback
   * manual</a> for further information about this appender.
   * </p>
   *
   * @param <E>
   * @author Ceki G&uuml;lc&uuml;
   * @author Torsten Juergeleit
   * @since 1.0.4
   */
  public class AsyncAppenderBase<E> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {
  
      AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<E>();
      BlockingQueue<E> blockingQueue;
  
      /**
       * The default buffer size.
       */
      public static final int DEFAULT_QUEUE_SIZE = 256;
      int queueSize = DEFAULT_QUEUE_SIZE;
  
      int appenderCount = 0;
  
      static final int UNDEFINED = -1;
      int discardingThreshold = UNDEFINED;
      boolean neverBlock = false;
  
      Worker worker = new Worker();
  
      /**
       * The default maximum queue flush time allowed during appender stop. If the
       * worker takes longer than this time it will exit, discarding any remaining
       * items in the queue
       */
      public static final int DEFAULT_MAX_FLUSH_TIME = 1000;
      int maxFlushTime = DEFAULT_MAX_FLUSH_TIME;
  
      /**
       * Is the eventObject passed as parameter discardable? The base class's
       * implementation of this method always returns 'false' but sub-classes may (and
       * do) override this method.
       * <p>
       * Note that only if the buffer is nearly full are events discarded. Otherwise,
       * when the buffer is "not full" all events are logged.
       *
       * @param eventObject
       * @return - true if the event can be discarded, false otherwise
       */
      protected boolean isDiscardable(E eventObject) {
          return false;
      }
  
      /**
       * Pre-process the event prior to queueing. The base class does no
       * pre-processing but subclasses can override this behavior.
       *
       * @param eventObject
       */
      protected void preprocess(E eventObject) {
      }
  
      @Override
      public void start() {
          if (isStarted())
             return;
         if (appenderCount == 0) {
             addError("No attached appenders found.");
             return;
         }
         if (queueSize < 1) {
             addError("Invalid queue size [" + queueSize + "]");
             return;
         }
         blockingQueue = new ArrayBlockingQueue<E>(queueSize);
 
         if (discardingThreshold == UNDEFINED)
             discardingThreshold = queueSize / 5;
         addInfo("Setting discardingThreshold to " + discardingThreshold);
         worker.setDaemon(true);
         worker.setName("AsyncAppender-Worker-" + getName());
         // make sure this instance is marked as "started" before staring the worker
         // Thread
         super.start();
         worker.start();
     }
 
     @Override
     public void stop() {
         if (!isStarted())
             return;
 
         // mark this appender as stopped so that Worker can also processPriorToRemoval
         // if it is invoking
         // aii.appendLoopOnAppenders
         // and sub-appenders consume the interruption
         super.stop();
 
         // interrupt the worker thread so that it can terminate. Note that the
         // interruption can be consumed by sub-appenders
         worker.interrupt();
 
         InterruptUtil interruptUtil = new InterruptUtil(context);
 
         try {
             interruptUtil.maskInterruptFlag();
 
             worker.join(maxFlushTime);
 
             // check to see if the thread ended and if not add a warning message
             if (worker.isAlive()) {
                 addWarn("Max queue flush timeout (" + maxFlushTime + " ms) exceeded. Approximately "
                         + blockingQueue.size() + " queued events were possibly discarded.");
             } else {
                 addInfo("Queue flush finished successfully within timeout.");
             }
 
         } catch (InterruptedException e) {
             int remaining = blockingQueue.size();
             addError("Failed to join worker thread. " + remaining + " queued events may be discarded.", e);
         } finally {
             interruptUtil.unmaskInterruptFlag();
         }
     }
 
     @Override
     protected void append(E eventObject) {
         if (isQueueBelowDiscardingThreshold() && isDiscardable(eventObject)) {
             return;
         }
         preprocess(eventObject);
         put(eventObject);
     }
 
     private boolean isQueueBelowDiscardingThreshold() {
         return (blockingQueue.remainingCapacity() < discardingThreshold);
     }
 
     private void put(E eventObject) {
         if (neverBlock) {
             blockingQueue.offer(eventObject);
         } else {
             putUninterruptibly(eventObject);
         }
     }
 
     private void putUninterruptibly(E eventObject) {
         boolean interrupted = false;
         try {
             while (true) {
                 try {
                     blockingQueue.put(eventObject);
                     break;
                 } catch (InterruptedException e) {
                     interrupted = true;
                 }
             }
         } finally {
             if (interrupted) {
                 Thread.currentThread().interrupt();
             }
         }
     }
 
     public int getQueueSize() {
         return queueSize;
     }
 
     public void setQueueSize(int queueSize) {
         this.queueSize = queueSize;
     }
 
     public int getDiscardingThreshold() {
         return discardingThreshold;
     }
 
     public void setDiscardingThreshold(int discardingThreshold) {
         this.discardingThreshold = discardingThreshold;
     }
 
     public int getMaxFlushTime() {
         return maxFlushTime;
     }
 
     public void setMaxFlushTime(int maxFlushTime) {
         this.maxFlushTime = maxFlushTime;
     }
 
     /**
      * Returns the number of elements currently in the blocking queue.
      *
      * @return number of elements currently in the queue.
      */
     public int getNumberOfElementsInQueue() {
         return blockingQueue.size();
     }
 
     public void setNeverBlock(boolean neverBlock) {
         this.neverBlock = neverBlock;
     }
 
     public boolean isNeverBlock() {
         return neverBlock;
     }
 
     /**
      * The remaining capacity available in the blocking queue.
      * <p>
      * See also {@link java.util.concurrent.BlockingQueue#remainingCapacity()
      * BlockingQueue#remainingCapacity()}
      *
      * @return the remaining capacity
      * 
      */
     public int getRemainingCapacity() {
         return blockingQueue.remainingCapacity();
     }
 
     public void addAppender(Appender<E> newAppender) {
         if (appenderCount == 0) {
             appenderCount++;
             addInfo("Attaching appender named [" + newAppender.getName() + "] to AsyncAppender.");
             aai.addAppender(newAppender);
         } else {
             addWarn("One and only one appender may be attached to AsyncAppender.");
             addWarn("Ignoring additional appender named [" + newAppender.getName() + "]");
         }
     }
 
     public Iterator<Appender<E>> iteratorForAppenders() {
         return aai.iteratorForAppenders();
     }
 
     public Appender<E> getAppender(String name) {
         return aai.getAppender(name);
     }
 
     public boolean isAttached(Appender<E> eAppender) {
         return aai.isAttached(eAppender);
     }
 
     public void detachAndStopAllAppenders() {
         aai.detachAndStopAllAppenders();
     }
 
     public boolean detachAppender(Appender<E> eAppender) {
         return aai.detachAppender(eAppender);
     }
 
     public boolean detachAppender(String name) {
         return aai.detachAppender(name);
     }
 
     class Worker extends Thread {
 
         public void run() {
             AsyncAppenderBase<E> parent = AsyncAppenderBase.this;
             AppenderAttachableImpl<E> aai = parent.aai;
 
             // loop while the parent is started
             while (parent.isStarted()) {
                 try {
                     List<E> elements = new ArrayList<E>();
                     E e0 = parent.blockingQueue.take();
                     elements.add(e0);
                     parent.blockingQueue.drainTo(elements);
                     for (E e : elements) {
                         aai.appendLoopOnAppenders(e);
                     }
                 } catch (InterruptedException e1) {
                     // exit if interrupted
                     break;
                 }
             }
 
             addInfo("Worker thread will flush remaining events before exiting. ");
 
             for (E e : parent.blockingQueue) {
                 aai.appendLoopOnAppenders(e);
                 parent.blockingQueue.remove(e);
             }
 
             aai.detachAndStopAllAppenders();
         }
     }
 }