package com.sonymusic.delphi.log4j;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.helpers.LogLog;
import org.apache.log4j.spi.LoggingEvent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * An asynchronous appender that uses an LMAX Disruptor {@link RingBuffer} as the interthread data exchange mechanism.
 * <p>
 * See the <a href="https://lmax-exchange.github.io/disruptor/">LMAX Disruptor documentation</a>
 * for more information about the advantages of using a {@link RingBuffer} over a {@link BlockingQueue}.
 * <p>
 * <p>
 * This appender will never block the logging thread, since it uses
 * {@link RingBuffer#tryPublishEvent(EventTranslatorOneArg, Object)}
 * to publish events (rather than {@link RingBuffer#publishEvent(EventTranslatorOneArg, Object)}).
 * <p>
 * <p>
 * If the RingBuffer is full, and the event cannot be published,
 * the event will be dropped.  A warning message will be logged to
 * logback's context every {@link #droppedWarnFrequency} consecutive dropped events.
 * <p>
 * <p>
 * A single handler thread will be used to handle the actual handling of the event.
 * <p>
 * <p>
 * Subclasses are required to set the {@link #eventHandler} to define
 * the logic that executes in the handler thread.
 * <p>
 * <p>
 * By default, child threads created by this appender will be daemon threads,
 * and therefore allow the JVM to exit gracefully without
 * needing to explicitly shut down the appender.
 * Note that in this case, it is possible for appended log events to not
 * be handled (if the child thread has not had a chance to process them yet).
 * <p>
 * By setting {@link #setUseDaemonThread(boolean)} to false, you can change this behavior.
 * When false, child threads created by this appender will not be daemon threads,
 * and therefore will prevent the JVM from shutting down
 * until the appender is explicitly shut down.
 * Set this to false if you want to ensure that every log event
 * prior to shutdown is handled.
 */
public abstract class AsyncDisruptorAppender extends AppenderSkeleton {

    public static final String THREAD_NAME_FORMAT = "datadog-appender-%1$s";

    public static final int DEFAULT_RING_BUFFER_SIZE = 8192;
    public static final int DEFAULT_DROPPED_WARN_FREQUENCY = 1000;

    private static final RingBufferFullException RING_BUFFER_FULL_EXCEPTION = new RingBufferFullException();

    static {
        RING_BUFFER_FULL_EXCEPTION.setStackTrace(new StackTraceElement[]{
                new StackTraceElement(AsyncDisruptorAppender.class.getName(), "append(..)", null, -1)
        });
    }

    /**
     * The size of the {@link RingBuffer}.
     * Defaults to {@value #DEFAULT_RING_BUFFER_SIZE}.
     * If the handler thread is not as fast as the producing threads,
     * then the {@link RingBuffer} will eventually fill up,
     * at which point events will be dropped.
     * <p>
     * Must be a positive power of 2.
     */
    private int ringBufferSize = DEFAULT_RING_BUFFER_SIZE;

    /**
     * When true, child threads created by this appender will be daemon threads,
     * and therefore allow the JVM to exit gracefully without
     * needing to explicitly shut down the appender.
     * Note that in this case, it is possible for log events to not
     * be handled.
     * <p>
     * <p>
     * When false, child threads created by this appender will not be daemon threads,
     * and therefore will prevent the JVM from shutting down
     * until the appender is explicitly shut down.
     * Set this to false if you want to ensure that every log event
     * prior to shutdown is handled.
     */
    private boolean useDaemonThread = true;

    /**
     * For every droppedWarnFrequency consecutive dropped events, log a warning.
     * Defaults to {@value #DEFAULT_DROPPED_WARN_FREQUENCY}.
     */
    private int droppedWarnFrequency = DEFAULT_DROPPED_WARN_FREQUENCY;

    /**
     * The {@link Disruptor} containing the {@link RingBuffer} onto
     * which to publish events.
     */
    private Disruptor<LogEvent> disruptor;

    /**
     * Sets the {@link LogEvent#event} to the logback Event.
     * Used when publishing events to the {@link RingBuffer}.
     */
    private EventTranslatorOneArg<LogEvent, LoggingEvent> eventTranslator = new LogEventTranslator();

    /**
     * Used by the handler thread to process the event.
     */
    private EventHandler<LogEvent> eventHandler;

    /**
     * Defines what happens when there is an exception during
     * {@link RingBuffer} processing.
     */
    private ExceptionHandler<LogEvent> exceptionHandler = new LogEventExceptionHandler();

    /**
     * Consecutive number of dropped events.
     */
    private final AtomicLong consecutiveDroppedCount = new AtomicLong();

    private boolean locationInfo = false;

    public AsyncDisruptorAppender() {
        Runtime.getRuntime().addShutdownHook(new Thread(this::close));
    }

    /**
     * Event wrapper object used for each element of the {@link RingBuffer}.
     */
    protected static class LogEvent {

        /**
         * The logging event.
         */
        public volatile LoggingEvent event;
    }

    /**
     * Sets the {@link LogEvent#event} to the logback Event.
     * Used when publishing events to the {@link RingBuffer}.
     */
    protected class LogEventTranslator implements EventTranslatorOneArg<LogEvent, LoggingEvent> {

        @Override
        public void translateTo(LogEvent logEvent, long sequence, LoggingEvent loggingEvent) {
            if (loggingEvent != null) {
                // Set the NDC and thread name for the calling thread as these
                // LoggingEvent fields were not set at event creation time.
                loggingEvent.getNDC();
                loggingEvent.getThreadName();
                // Get a copy of this thread's MDC.
                loggingEvent.getMDCCopy();
                if (locationInfo) {
                    loggingEvent.getLocationInformation();
                }
                loggingEvent.getRenderedMessage();
                loggingEvent.getThrowableStrRep();
            }

            // put logging event to disruptor event
            logEvent.event = loggingEvent;
        }
    }

    /**
     * Defines what happens when there is an exception during
     * {@link RingBuffer} processing.
     * <p>
     * Currently, just logs to the logback context.
     */
    private static class LogEventExceptionHandler implements ExceptionHandler<LogEvent> {

        @Override
        public void handleEventException(Throwable ex, long sequence, LogEvent event) {
            LogLog.error("Unable to process event: " + ex.getMessage(), ex);
        }

        @Override
        public void handleOnStartException(Throwable ex) {
            LogLog.error("Unable start disruptor", ex);
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {
            LogLog.error("Unable shutdown disruptor", ex);
        }
    }

    /**
     * Clears the event after a delegate event handler has processed the event,
     * so that the event can be garbage collected.
     */
    private static class EventClearingEventHandler implements EventHandler<LogEvent>, LifecycleAware {

        private final EventHandler<LogEvent> delegate;

        public EventClearingEventHandler(EventHandler<LogEvent> delegate) {
            super();
            this.delegate = delegate;
        }

        @Override
        public void onEvent(LogEvent event, long sequence, boolean endOfBatch) throws Exception {
            try {
                delegate.onEvent(event, sequence, endOfBatch);
            } finally {
                // Clear the event so that it can be garbage collected.
                event.event = null;
            }
        }

        @Override
        public void onStart() {
            if (delegate instanceof LifecycleAware) {
                ((LifecycleAware) delegate).onStart();
            }
        }

        @Override
        public void onShutdown() {
            if (delegate instanceof LifecycleAware) {
                ((LifecycleAware) delegate).onShutdown();
            }
        }

    }

    @Override
    public void activateOptions() {
        if (eventHandler == null) {
            LogLog.error("No eventHandler was configured for appender " + name + ".");
            return;
        }

        disruptor = new Disruptor<>(
                LogEvent::new,
                ringBufferSize,
                new WorkerThreadFactory(() -> String.format(THREAD_NAME_FORMAT, getName()), useDaemonThread)
        );

        // Define the exceptionHandler first, so that it applies to all future eventHandlers.
        disruptor.setDefaultExceptionHandler(exceptionHandler);
        disruptor.handleEventsWith(new EventClearingEventHandler(eventHandler));

        disruptor.start();
        super.activateOptions();

        LogLog.debug(String.format("Appender %s started", getClass().getName()));
    }

    @Override
    public void close() {
        // Don't allow any more events to be appended.
        try {
            this.disruptor.shutdown(1, TimeUnit.MINUTES);
        } catch (TimeoutException e) {
            LogLog.warn("Some queued events have not been logged due to requested shutdown");
        }

        LogLog.debug(String.format("Appender %s stopped", getClass().getName()));
    }

    @Override
    protected void append(LoggingEvent event) {
        long startTime = System.nanoTime();

        if (!this.disruptor.getRingBuffer().tryPublishEvent(this.eventTranslator, event)) {
            long consecutiveDropped = this.consecutiveDroppedCount.incrementAndGet();
            if ((consecutiveDropped) % this.droppedWarnFrequency == 1) {
                LogLog.warn("Dropped " + consecutiveDropped + " events (and counting...) due to ring buffer at max capacity [" + this.ringBufferSize + "]");
            }
            LogLog.error("Appending logging event failed", RING_BUFFER_FULL_EXCEPTION);
        } else {
            long endTime = System.nanoTime();
            long consecutiveDropped = this.consecutiveDroppedCount.get();
            if (consecutiveDropped != 0 && this.consecutiveDroppedCount.compareAndSet(consecutiveDropped, 0L)) {
                LogLog.warn("Dropped " + consecutiveDropped + " total events due to ring buffer at max capacity [" + this.ringBufferSize + "]");
            }
            LogLog.debug(String.format("Logging event successfully appended (%s nanos).", endTime - startTime));
        }
    }

    protected Disruptor<LogEvent> getDisruptor() {
        return disruptor;
    }

    public EventTranslatorOneArg<LogEvent, LoggingEvent> getEventTranslator() {
        return eventTranslator;
    }

    public void setRingBufferSize(int ringBufferSize) {
        this.ringBufferSize = ringBufferSize;
    }

    public void setDroppedWarnFrequency(int droppedWarnFrequency) {
        this.droppedWarnFrequency = droppedWarnFrequency;
    }

    protected void setEventHandler(EventHandler<LogEvent> eventHandler) {
        this.eventHandler = eventHandler;
    }

    public void setUseDaemonThread(boolean useDaemonThread) {
        this.useDaemonThread = useDaemonThread;
    }

    public boolean isUseDaemonThread() {
        return useDaemonThread;
    }

    public void setLocationInfo(boolean locationInfo) {
        this.locationInfo = locationInfo;
    }
}