package com.sonymusic.delphi.log4j;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.LifecycleAware;
import org.apache.log4j.helpers.LogLog;
import org.apache.log4j.spi.LoggingEvent;

import javax.net.SocketFactory;
import javax.net.ssl.SSLSocket;
import java.io.*;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * An {@link AsyncDisruptorAppender} appender that writes events to a TCP {@link Socket} outputStream.
 */
public abstract class AbstractTcpSocketAppender extends AsyncDisruptorAppender {

    public static final String THREAD_NAME_FORMAT = "socket-monitor-%1$s-%2$s:%3$d";

    /**
     * Executor service for scheduling periodical events
     */
    private ScheduledExecutorService scheduledExecutorService;

    /**
     * The default reconnection delay (30000 milliseconds or 30 seconds).
     */
    public static final int DEFAULT_RECONNECTION_DELAY = 30000;

    /**
     * The default write timeout in milliseconds (0 means no write timeout).
     */
    public static final int DEFAULT_WRITE_TIMEOUT = 0;

    /**
     * Default timeout when waiting for the remote server to accept our
     * connection.
     */
    public static final int DEFAULT_CONNECTION_TIMEOUT = 5000;

    public static final int DEFAULT_WRITE_BUFFER_SIZE = 8192;

    private static final NotConnectedException NOT_CONNECTED_EXCEPTION = new NotConnectedException();
    private static final ShutdownInProgressException SHUTDOWN_IN_PROGRESS_EXCEPTION = new ShutdownInProgressException();

    static {
        NOT_CONNECTED_EXCEPTION.setStackTrace(new StackTraceElement[]{
                new StackTraceElement(AbstractTcpSocketAppender.TcpSendingEventHandler.class.getName(), "onEvent(..)", null, -1)
        });

        SHUTDOWN_IN_PROGRESS_EXCEPTION.setStackTrace(new StackTraceElement[]{
                new StackTraceElement(AbstractTcpSocketAppender.TcpSendingEventHandler.class.getName(), "onEvent(..)", null, -1)
        });
    }

    /**
     * The host to which to connect and send events
     */
    private String host;

    /**
     * The TCP port on the host to which to connect and send events
     */
    private int port;

    /**
     * When connected, this is the connected destination address.
     * When not connected, this is null.
     */
    private volatile InetSocketAddress connectedDestination;

    /**
     * Time period for which to wait after a connection fails to a specific destination
     * before attempting to reconnect to that destination.
     * Default is {@value #DEFAULT_RECONNECTION_DELAY} milliseconds.
     */
    private Duration reconnectionDelay = Duration.ofMillis(DEFAULT_RECONNECTION_DELAY);

    /**
     * Socket connection timeout in milliseconds.
     */
    private int acceptConnectionTimeout = DEFAULT_CONNECTION_TIMEOUT;

    /**
     * The number of bytes available in the write buffer.
     * Defaults to {@value #DEFAULT_WRITE_BUFFER_SIZE}
     * <p>
     * If less than or equal to zero, buffering the output stream will be disabled.
     * If buffering is disabled, the writer thread can slow down, but
     * it will also can prevent dropping events in the buffer on flaky connections.
     */
    private int writeBufferSize = DEFAULT_WRITE_BUFFER_SIZE;

    /**
     * Used to create client {@link Socket}s to which to communicate.
     * <p>
     * If set prior to startup, it will be used.
     */
    private SocketFactory socketFactory;

    /**
     * If this duration elapses without an event being sent,
     * then the {@link #keepAliveMessage} will be sent to the socket in
     * order to keep the connection alive.
     * <p>
     * When null (the default), no keepAlive messages will be sent.
     */
    private Duration keepAliveDuration;

    /**
     * Message to send for keeping the connection alive
     * if {@link #keepAliveDuration} is non-null.
     */
    private KeepAliveMessage keepAliveMessage = KeepAliveMessage.SYSTEM;

    /**
     * The charset to use when writing the {@link #keepAliveMessage}.
     * Defaults to UTF-8.
     */
    private Charset keepAliveCharset = StandardCharsets.UTF_8;

    /**
     * The {@link #keepAliveMessage} translated to bytes using the {@link #keepAliveCharset}.
     * Populated at startup time.
     */
    private byte[] keepAliveBytes;

    /**
     * Time period for which to wait for a write to complete before timing out
     * and attempting to reconnect to that destination.
     * Zero (the default) means no write timeout.
     *
     * <p>Used to detect connections where the receiver stops reading.</p>
     *
     * <p>Note that since a blocking java socket output stream
     * does not have a concept of a write timeout,
     * a task will be scheduled on the {@link ScheduledExecutorService}
     * with the same frequency as the write timeout
     * in order to detect stuck writes.
     * It is recommended to use longer write timeouts (e.g. &gt; 30s, or minutes),
     * rather than short write timeouts, so that this task does not execute too frequently.
     * Also, this approach means that it could take up to two times the write timeout
     * before a write timeout is detected.</p>
     */
    private Duration writeTimeout = Duration.ofMillis(DEFAULT_WRITE_TIMEOUT);

    /**
     * Used to signal the socket reconnect thread that the shutdown has occurred.
     * The latch will be non-zero when started, and zero when shutdown.
     */
    private volatile CountDownLatch shutdownLatch;

    /**
     * Event handler responsible for performing the TCP transmission.
     */
    private class TcpSendingEventHandler implements EventHandler<LogEvent>, LifecycleAware {

        /**
         * Max number of consecutive failed connection attempts for which
         * logback status messages will be logged.
         * <p>
         * After this many failed attempts, reconnection will still
         * be attempted, but failures will not be logged again
         * (until after the connection is successful, and then fails again.)
         */
        private static final int MAX_REPEAT_CONNECTION_ERROR_LOG = 5;

        /**
         * Number of times we try to write an event before it is discarded.
         * Between each attempt, the socket will be reconnected.
         */
        private static final int MAX_REPEAT_WRITE_ATTEMPTS = 5;

        /**
         * The destination socket to which to send events.
         */
        private volatile Socket socket;

        /**
         * The destination output stream to which to send events.
         * If {@link AbstractTcpSocketAppender#writeBufferSize} is greater than zero, this will be a buffered wrapper of the socket output stream.
         * Otherwise, it will be the socket output stream.
         */
        private volatile OutputStream outputStream;

        /**
         * Time at which the last event send was started (e.g. before write/flush).
         * Used to detect write timeouts.
         */
        private volatile long lastSendStartNanoTime;
        /**
         * Time at which the last event send was completed (e.g. after write/flush).
         * Used to calculate if a keep alive message
         * needs to be scheduled/sent.
         */
        private volatile long lastSendEndNanoTime;

        /**
         * The most recent time that a connection to each destination was attempted.
         */
        private long destinationAttemptStartTime;

        /**
         * Future for the currently scheduled {@link #keepAliveRunnable}.
         */
        private ScheduledFuture<?> keepAliveFuture;

        /**
         * See {@link KeepAliveRunnable}.
         * Initialized on startup if keep alive is enabled.
         */
        private KeepAliveRunnable keepAliveRunnable;

        /**
         * Future for the currently scheduled {@link #writeTimeoutRunnable}.
         */
        private ScheduledFuture<?> writeTimeoutFuture;

        /**
         * See {@link WriteTimeoutRunnable}.
         * Initialized on startup if write timeout is enabled.
         */
        private WriteTimeoutRunnable writeTimeoutRunnable;

        /**
         * See {@link ReaderCallable}.
         * Initialized when a socket is opened.
         */
        private Future<?> readerFuture;

        /**
         * When run, if the {@link AbstractTcpSocketAppender#keepAliveDuration}
         * has elapsed since the last event was sent,
         * then this runnable will publish a keepAlive event to the ringBuffer.
         * <p>
         * The runnable will reschedule itself to execute in the future
         * after the calculated {@link AbstractTcpSocketAppender#keepAliveDuration}
         * from the last sent event using {@link TcpSendingEventHandler#scheduleKeepAlive(long)}.
         * <p>
         * When the keepAlive event is processed by the event handler,
         * if the {@link AbstractTcpSocketAppender#keepAliveDuration}
         * has elapsed since the last event was sent,
         * then the event handler will send the {@link AbstractTcpSocketAppender#keepAliveMessage}
         * to the socket outputstream.
         */
        private class KeepAliveRunnable implements Runnable {

            @Override
            public void run() {
                long lastSendEnd = lastSendEndNanoTime;
                long currentNanoTime = System.nanoTime();
                if (hasKeepAliveDurationElapsed(lastSendEnd, currentNanoTime)) {
                    /*
                     * Publish a keep alive message to the RingBuffer.
                     *
                     * A null event indicates that this is a keep alive message.
                     *
                     * Use tryPublishEvent instead of publishEvent, because if the ring buffer is full,
                     * there's really no need to send a keep alive, since
                     * there are other messages waiting to be sent.
                     */
                    getDisruptor().getRingBuffer().tryPublishEvent(getEventTranslator(), null);
                    scheduleKeepAlive(currentNanoTime);
                } else {
                    scheduleKeepAlive(lastSendEnd);
                }
            }
        }

        /**
         * Keeps reading the {@link ReaderCallable#inputStream} until the
         * end of the stream is reached.
         * <p>
         * This helps pro-actively detect server-side socket disconnections,
         * specifically in the case of Amazon's Elastic Load Balancers (ELB).
         */
        private class ReaderCallable implements Callable<Void> {

            private final InputStream inputStream;

            public ReaderCallable(InputStream inputStream) {
                super();
                this.inputStream = inputStream;
            }

            @Override
            public Void call() throws Exception {
                Thread.currentThread().setName(calculateThreadPrefix());
                try {
                    while (true) {
                        try {
                            if (inputStream.read() == -1) {
                                // End of stream reached, so we're done.
                                return null;
                            }
                        } catch (SocketTimeoutException e) {
                            // ignore, and try again
                        }
                    }
                } finally {
                    if (!Thread.currentThread().isInterrupted()) {
                        scheduledExecutorService.submit(() -> {
                            /*
                             * https://github.com/logstash/logstash-logback-encoder/issues/341
                             *
                             * Pro-actively trigger the event handler's onEvent method in the handler thread
                             * by publishing a null event (which usually means a keepAlive event).
                             *
                             * When onEvent handles the event in the handler thread,
                             * it will detect that readerFuture.isDone() and reopen the socket.
                             *
                             * Without this, onEvent would not be called until the next event,
                             * which might not occur for a while.
                             * So, this is really just an optimization to reopen the socket as soon as possible.
                             *
                             * We can't reopen the socket from this thread,
                             * since all socket open/close must be done from the event handler thread.
                             *
                             * There is a potential race condition here as well, since
                             * onEvent could be triggered before the readerFuture completes.
                             * We reduce (but not eliminate) the chance of that happening by
                             * scheduling this task on the executorService.
                             */
                            getDisruptor().getRingBuffer().tryPublishEvent(getEventTranslator(), null);
                        });
                    }
                }
            }

        }

        /**
         * Detects write timeouts by inspecting {@link #lastSendStartNanoTime} and {@link #lastSendEndNanoTime}
         */
        private class WriteTimeoutRunnable implements Runnable {

            /**
             * The lastSendStartNanoTime of the last detected timeout.
             * Used to ensure we only detect a write timeout for a single write once
             * (especially if the log rate is very low).
             */
            private volatile long lastDetectedStartNanoTime;

            @Override
            public void run() {
                long lastSendStart = lastSendStartNanoTime; // volatile read
                long lastSendEnd = lastSendEndNanoTime;     // volatile read

                /*
                 * A write is in progress if the start is greater than the end
                 */
                if (lastSendStart > lastSendEnd && lastSendStart != lastDetectedStartNanoTime) {

                    long elapsedSendTimeInMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastSendStart);
                    if (elapsedSendTimeInMillis > writeTimeout.toMillis()) {
                        lastDetectedStartNanoTime = lastSendStart;
                        LogLog.warn("Detected write timeout after " + elapsedSendTimeInMillis + "ms.  Write timeout=" + writeTimeout + ".  Closing socket to force reconnect");
                        closeSocket();
                    }
                }
            }
        }

        @Override
        public void onEvent(LogEvent logEvent, long sequence, boolean endOfBatch) {

            Exception sendFailureException = null;
            for (int i = 0; i < MAX_REPEAT_WRITE_ATTEMPTS; i++) {
                /*
                 * Save local references to the outputStream and socket
                 * in case the WriteTimeoutRunnable closes the socket.
                 */
                Socket socket = this.socket; // volatile read
                OutputStream outputStream = this.outputStream; // volatile read

                if (socket == null && (!isStarted() || Thread.currentThread().isInterrupted())) {
                    /*
                     * Handle shutdown in progress
                     *
                     * This will occur if shutdown occurred during reopen()
                     */
                    sendFailureException = SHUTDOWN_IN_PROGRESS_EXCEPTION;
                    break;
                }

                Future<?> readerFuture = this.readerFuture;  // volatile read
                if (readerFuture.isDone() || socket == null) {
                    /*
                     * If readerFuture.isDone(), then the destination has shut down its output (our input),
                     * and the destination is probably no longer listening to its input (our output).
                     * This will be the case for Amazon's Elastic Load Balancers (ELB)
                     * when an instance behind the ELB becomes unhealthy while we're connected to it.
                     *
                     * If socket == null here, it means that a write timed out,
                     * and the socket was closed by the WriteTimeoutRunnable.
                     *
                     * Therefore, attempt reconnection.
                     */
                    LogLog.debug("Destination terminated the connection. Reconnecting.");
                    reopenSocket();
                    try {
                        readerFuture.get();
                        sendFailureException = NOT_CONNECTED_EXCEPTION;
                    } catch (Exception e) {
                        sendFailureException = e;
                    }
                    continue;
                }
                try {
                    writeEvent(outputStream, logEvent, endOfBatch);
                    return;
                } catch (Exception e) {
                    sendFailureException = e;
                    LogLog.warn("Unable to send event: " + e.getMessage() + " Reconnecting.", e);
                    /*
                     * Need to re-open the socket in case of IOExceptions.
                     *
                     * Reopening the socket probably won't help other exceptions
                     * (like NullPointerExceptions),
                     * but we're doing so anyway, just in case.
                     */
                    reopenSocket();
                }
            }

            LogLog.error("Unable to send event to TCP socket", sendFailureException);
        }

        private void writeEvent(OutputStream outputStream, LogEvent logEvent, boolean endOfBatch) throws IOException {

            long startNanoTime = System.nanoTime();
            lastSendStartNanoTime = startNanoTime;
            /*
             * A null event indicates that this is a keep alive message,
             * or an event sent from the ReaderCallable.
             */
            if (logEvent.event != null) {
                /*
                 * This is a standard (non-keepAlive) event.
                 * Therefore, we need to send the event.
                 */
                outputStream.write(encodeLoggingEvent(logEvent.event));
            } else if (hasKeepAliveDurationElapsed(lastSendEndNanoTime, startNanoTime)) {
                /*
                 * This is a keep alive event, and the keepAliveDuration has passed,
                 * Therefore, we need to send the keepAliveMessage.
                 */
                outputStream.write(keepAliveBytes);
            }
            if (endOfBatch) {
                outputStream.flush();
            }
            lastSendEndNanoTime = System.nanoTime();

            LogLog.debug("Logging event sent to socket");
        }

        private boolean hasKeepAliveDurationElapsed(long lastSentNanoTime, long currentNanoTime) {
            return isKeepAliveEnabled()
                    && lastSentNanoTime + TimeUnit.MILLISECONDS.toNanos(keepAliveDuration.toMillis()) < currentNanoTime;
        }

        @Override
        public void onStart() {
            // Core size to handle the reader thread
            int threadPoolCoreSize = 1;

            // Increase the core size to handle the keep alive thread
            if (isKeepAliveEnabled()) {
                threadPoolCoreSize++;
            }

            // Increase the core size to handle the write timeout detection thread
            if (isWriteTimeoutEnabled()) {
                threadPoolCoreSize++;
            }

            scheduledExecutorService = Executors.newScheduledThreadPool(
                    threadPoolCoreSize,
                    new WorkerThreadFactory(AbstractTcpSocketAppender.this::calculateThreadPrefix, isUseDaemonThread())
            );

            openSocket();
            scheduleKeepAlive(System.nanoTime());
            scheduleWriteTimeout();
        }

        @Override
        public void onShutdown() {
            unscheduleWriteTimeout();
            unscheduleKeepAlive();
            closeSocket();

            scheduledExecutorService.shutdown();
            try {
                if (!scheduledExecutorService.awaitTermination(1, TimeUnit.MINUTES)) {
                    LogLog.warn("Some queued events have not been logged due to requested shutdown");
                }
            } catch (InterruptedException e) {
                LogLog.warn("Some queued events have not been logged due to requested shutdown", e);
            }
        }

        private synchronized void reopenSocket() {
            closeSocket();
            openSocket();
        }

        /**
         * Repeatedly tries to open a socket until it is successful,
         * or the hander is stopped, or the handler thread is interrupted.
         * <p>
         * If the socket is non-null when this method returns,
         * then it should be able to be used to send.
         */
        private synchronized void openSocket() {
            int errorCount = 0;
            while (isStarted() && !Thread.currentThread().isInterrupted()) {
                long startWallTime = System.currentTimeMillis();
                Socket tempSocket = null;
                OutputStream tempOutputStream = null;

                try {
                    /*
                     * Delay the connection attempt if the last attempt to the selected destination
                     * was less than the reconnectionDelay.
                     */
                    final long millisSinceLastAttempt = startWallTime - destinationAttemptStartTime;
                    if (millisSinceLastAttempt < reconnectionDelay.toMillis()) {
                        final long sleepTime = reconnectionDelay.toMillis() - millisSinceLastAttempt;
                        if (errorCount < MAX_REPEAT_CONNECTION_ERROR_LOG) {
                            LogLog.warn("Waiting " + sleepTime + "ms before attempting reconnection.");
                        }
                        try {
                            shutdownLatch.await(sleepTime, TimeUnit.MILLISECONDS);
                            if (!isStarted()) {
                                return;
                            }
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                            LogLog.warn("Connection interrupted. Will no longer attempt reconnection.");
                            return;
                        }
                        // reset the start time to be after the wait period.
                        startWallTime = System.currentTimeMillis();
                    }
                    destinationAttemptStartTime = startWallTime;

                    /*
                     * Set the SO_TIMEOUT so that SSL handshakes will timeout if they take too long.
                     *
                     * Note that SO_TIMEOUT only applies to reads (which occur during the handshake process).
                     */
                    tempSocket = socketFactory.createSocket();
                    tempSocket.setSoTimeout(acceptConnectionTimeout);
                    /*
                     * currentDestination is unresolved, so a new InetSocketAddress
                     * must be created to resolve the hostname.
                     */
                    InetSocketAddress destination = new InetSocketAddress(host, port);
                    tempSocket.connect(destination, acceptConnectionTimeout);

                    /*
                     * Trigger SSL handshake immediately and declare the socket unconnected if it fails
                     */
                    if (tempSocket instanceof SSLSocket) {
                        ((SSLSocket) tempSocket).startHandshake();
                    }

                    /*
                     * Issue #218, make buffering the output stream optional.
                     */
                    tempOutputStream = writeBufferSize > 0
                            ? new BufferedOutputStream(tempSocket.getOutputStream(), writeBufferSize)
                            : tempSocket.getOutputStream();

                    socket = tempSocket;
                    outputStream = tempOutputStream;

                    connectedDestination = destination;

                    ReaderCallable readerCallable = new ReaderCallable(tempSocket.getInputStream());
                    readerFuture = scheduledExecutorService.submit(readerCallable);

                    LogLog.debug("Connection opened.");
                    return;

                } catch (Exception e) {
                    closeQuietly(tempOutputStream);
                    closeQuietly(tempSocket);

                    // Avoid spamming status messages by checking the MAX_REPEAT_CONNECTION_ERROR_LOG.
                    if (errorCount++ < MAX_REPEAT_CONNECTION_ERROR_LOG) {
                        LogLog.warn("Connection failed.", e);
                    }
                }
            }
        }

        private synchronized void closeSocket() {
            connectedDestination = null;

            closeQuietly(outputStream);
            outputStream = null;

            closeQuietly(socket);
            socket = null;

            LogLog.debug("Socket closed.");

            if (this.readerFuture != null) {
                // This shouldn't be necessary, since closing the socket
                // should cause the read() call to throw an exception.
                // But cancel it anyway to be extra-safe.
                this.readerFuture.cancel(true);
            }
        }

        private synchronized void scheduleKeepAlive(long basedOnNanoTime) {
            if (isKeepAliveEnabled() && !Thread.currentThread().isInterrupted()) {
                if (keepAliveRunnable == null) {
                    keepAliveRunnable = new KeepAliveRunnable();
                }
                long delay = TimeUnit.MILLISECONDS.toNanos(keepAliveDuration.toMillis()) - (System.nanoTime() - basedOnNanoTime);
                try {
                    keepAliveFuture = scheduledExecutorService.schedule(
                            keepAliveRunnable,
                            delay,
                            TimeUnit.NANOSECONDS
                    );
                } catch (RejectedExecutionException e) {
                    // if scheduling failed, it means that the appender is shutting down.
                    keepAliveFuture = null;
                }
            }
        }

        private synchronized void unscheduleKeepAlive() {
            if (keepAliveFuture != null) {
                keepAliveFuture.cancel(true);
                try {
                    keepAliveFuture.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    // ignore
                } catch (Exception e) {
                    // ignore
                }
            }
        }

        private synchronized void scheduleWriteTimeout() {
            if (isWriteTimeoutEnabled() && !Thread.currentThread().isInterrupted()) {
                if (writeTimeoutRunnable == null) {
                    writeTimeoutRunnable = new WriteTimeoutRunnable();
                }
                long delay = writeTimeout.toMillis();
                try {
                    writeTimeoutFuture = scheduledExecutorService.scheduleWithFixedDelay(
                            writeTimeoutRunnable,
                            delay,
                            delay,
                            TimeUnit.MILLISECONDS
                    );
                } catch (RejectedExecutionException e) {
                    // if scheduling failed, it means that the appender is shutting down.
                    writeTimeoutFuture = null;
                }
            }
        }

        private synchronized void unscheduleWriteTimeout() {
            if (writeTimeoutFuture != null) {
                writeTimeoutFuture.cancel(true);
                try {
                    writeTimeoutFuture.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    // ignore
                } catch (Exception e) {
                    // ignore
                }
            }
        }
    }

    public AbstractTcpSocketAppender() {
        setEventHandler(new TcpSendingEventHandler());
    }

    public boolean isStarted() {
        CountDownLatch latch = this.shutdownLatch;
        return latch != null && latch.getCount() != 0;
    }


    @Override
    public void activateOptions() {
        if (isStarted()) {
            return;
        }

        // Create socket factory if not created yet
        if (socketFactory == null) {
            socketFactory = SocketFactory.getDefault();
        }

        if (keepAliveMessage != null && keepAliveCharset != null) {
            keepAliveBytes = keepAliveMessage.toString().getBytes(keepAliveCharset);
        }

        this.shutdownLatch = new CountDownLatch(1);
        super.activateOptions();
    }

    @Override
    public void close() {
        if (!isStarted()) {
            return;
        }

        // Stop waiting to reconnect (if reconnect logic is currently waiting)
        this.shutdownLatch.countDown();
        super.close();
    }

    /**
     * The host to which to connect and send events
     */
    public void setHost(String host) {
        this.host = host;
    }

    /**
     * The TCP port on the host to which to connect and send events
     */
    public void setPort(int port) {
        this.port = port;
    }

    private void closeQuietly(Closeable closeable) {
        if (closeable == null) {
            return;
        }

        try {
            closeable.close();
        } catch (Exception ex) {
            // do nothing
        }
    }

    protected String calculateThreadPrefix() {
        List<Object> threadNameFormatParams = getThreadNameFormatParams();
        return String.format(THREAD_NAME_FORMAT, threadNameFormatParams.toArray(new Object[0]));
    }

    protected List<Object> getThreadNameFormatParams() {
        List<Object> threadNameFormatParams = new ArrayList<>(3);
        threadNameFormatParams.add(getName());
        threadNameFormatParams.add(connectedDestination.getHostString());
        threadNameFormatParams.add(connectedDestination.getPort());
        return threadNameFormatParams;
    }

    /**
     * Time period for which to wait after failing to connect to all servers,
     * before attempting to reconnect.
     * Default is {@value #DEFAULT_RECONNECTION_DELAY} milliseconds.
     */
    public void setReconnectionDelay(String duration) {
        Duration delay = Duration.parse(duration);

        if (delay.toMillis() <= 0) {
            throw new IllegalArgumentException("reconnectionDelay must be > 0");
        }

        this.reconnectionDelay = delay;
    }

    /**
     * Socket connection timeout in milliseconds.
     */
    void setAcceptConnectionTimeout(int acceptConnectionTimeout) {
        this.acceptConnectionTimeout = acceptConnectionTimeout;
    }

    /**
     * The number of bytes available in the write buffer.
     * Defaults to {@value #DEFAULT_WRITE_BUFFER_SIZE}
     * <p>
     * If less than or equal to zero, buffering the output stream will be disabled.
     * If buffering is disabled, the writer thread can slow down, but
     * it will also can prevent dropping events in the buffer on flaky connections.
     */
    public void setWriteBufferSize(int writeBufferSize) {
        this.writeBufferSize = writeBufferSize;
    }

    /**
     * Sets the maximum number of events in the queue. Once the queue is full
     * additional events will be dropped.
     *
     * <p>
     * Must be a positive power of 2.
     *
     * @param queueSize the maximum number of entries in the queue.
     */
    public void setQueueSize(int queueSize) {
        setRingBufferSize(queueSize);
    }

    /**
     * If this duration elapses without an event being sent,
     * then the {@link #keepAliveMessage} will be sent to the socket in
     * order to keep the connection alive.
     * <p>
     * When null, no keepAlive messages will be sent.
     */
    public void setKeepAliveDuration(String keepAliveDuration) {
        this.keepAliveDuration = Duration.parse(keepAliveDuration);
    }

    /**
     * Message to send for keeping the connection alive
     * if {@link #keepAliveDuration} is non-null.
     * <p>
     * The following values have special meaning:
     * <ul>
     * <li><tt>null</tt> or empty string = no keep alive.</li>
     * <li>"<tt>SYSTEM</tt>" = operating system new line (default).</li>
     * <li>"<tt>UNIX</tt>" = unix line ending (\n).</li>
     * <li>"<tt>WINDOWS</tt>" = windows line ending (\r\n).</li>
     * </ul>
     * <p>
     * Any other value will be used as-is.
     */
    public void setKeepAliveMessage(String keepAliveMessage) {
        this.keepAliveMessage = KeepAliveMessage.valueOf(keepAliveMessage);
    }

    public boolean isKeepAliveEnabled() {
        return keepAliveDuration != null && keepAliveMessage != null;
    }

    public boolean isWriteTimeoutEnabled() {
        return this.writeTimeout.toMillis() > 0;
    }

    /**
     * The charset to use when writing the {@link #keepAliveMessage}.
     * Defaults to UTF-8.
     */
    public void setKeepAliveCharset(Charset keepAliveCharset) {
        this.keepAliveCharset = keepAliveCharset;
    }

    /**
     * Sets the time period for which to wait for a write to complete before timing out
     * and attempting to reconnect to that destination.
     * Zero (the default) means no write timeout.
     *
     * <p>Used to detect connections where the receiver stops reading.</p>
     *
     * <p>Note that since a blocking java socket output stream
     * does not have a concept of a write timeout,
     * a task will be scheduled on the {@link ScheduledExecutorService}
     * with the same frequency as the write timeout
     * in order to detect stuck writes.
     * It is recommended to use longer write timeouts (e.g. &gt; 30s, or minutes),
     * rather than short write timeouts, so that this task does not execute too frequently.
     * Also, this approach means that it could take up to two times the write timeout
     * before a write timeout is detected.</p>
     */
    public void setWriteTimeout(Duration writeTimeout) {
        this.writeTimeout = writeTimeout == null
                ? Duration.ofMillis(DEFAULT_WRITE_TIMEOUT)
                : writeTimeout;
    }

    public abstract byte[] encodeLoggingEvent(LoggingEvent loggingEvent);
}