/*
 * Copyright 2013 TORCH UG
 *
 * This file is part of Graylog2.
 *
 * Graylog2 is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Graylog2 is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Graylog2.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.graylog2.logback.appender;

import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import com.google.common.base.Splitter;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.oio.OioClientSocketChannelFactory;
import org.slf4j.LoggerFactory;
import play.Application;
import play.Configuration;
import play.Plugin;

import java.net.InetSocketAddress;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static org.graylog2.logback.appender.GelfTcpAppender.DONT_SEND_TO_GRAYLOG2;

@SuppressWarnings("unused")
public class Graylog2Plugin extends Plugin {
    private static final org.slf4j.Logger log = LoggerFactory.getLogger(Graylog2Plugin.class);

    private ChannelFuture channelFuture;
    private GelfTcpAppender gelfAppender;
    private GelfAppenderHandler handler;

    private InetSocketAddress graylog2ServerAddress;
    private final Integer queueCapacity;
    private ClientBootstrap bootstrap;
    private Graylog2Plugin.ReconnectListener reconnectListener;

    public Graylog2Plugin(Application app) {
        final Configuration config = app.configuration();
        queueCapacity = config.getInt("graylog2.appender.queue-size", 512);

        final String entry = config.getString("graylog2.appender.host", "127.0.0.1:12201");
        final Iterable<String> parts = Splitter.on(':').trimResults().omitEmptyStrings().limit(2).split(entry.toString());
        final Iterator<String> it = parts.iterator();
        try {
            String host = it.next();
            String port = it.next();
            graylog2ServerAddress = new InetSocketAddress(host, Integer.valueOf(port));
        } catch (IllegalArgumentException | NoSuchElementException e) {
            log.error("Malformed graylog2.appender.hosts entry {}. Please specify them as 'host:port'!", entry);
        }

    }

    @Override
    public void onStart() {
        bootstrap = new ClientBootstrap(
                new OioClientSocketChannelFactory(
                        Executors.newCachedThreadPool()));

        reconnectListener = new ReconnectListener();

        handler = new GelfAppenderHandler(this, queueCapacity);

        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() throws Exception {
                return Channels.pipeline(handler);
            }
        });
        new Thread(new Runnable() {
            @Override
            public void run() {
                channelFuture = reconnect();
            }
        }).start();
        LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
        Logger rootLogger = lc.getLogger(Logger.ROOT_LOGGER_NAME);

        gelfAppender = new GelfTcpAppender(handler);
        gelfAppender.setContext(lc);
        gelfAppender.start();

        rootLogger.addAppender(gelfAppender);
    }

    public ChannelFuture reconnect() {
        if (log.isDebugEnabled()) {
            log.debug(DONT_SEND_TO_GRAYLOG2, "Reconnecting to graylog2 at {}", graylog2ServerAddress);
        }
        channelFuture = bootstrap.connect(graylog2ServerAddress);
        channelFuture.addListener(reconnectListener);
        return channelFuture;
    }

    @Override
    public void onStop() {
        if (gelfAppender != null) gelfAppender.stop();
        if (handler != null) handler.stop();
        if (channelFuture.getChannel()!= null) channelFuture.getChannel().close().awaitUninterruptibly();
    }

    private class ReconnectListener implements ChannelFutureListener {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                if (log.isDebugEnabled()) {
                    log.debug(DONT_SEND_TO_GRAYLOG2, "Could not connect. Retrying in 500 ms. Exception {}", future.getCause().getMessage());
                }
                // TODO implement back-off
                Thread.sleep(500);
                // could not connect, let's retry
                reconnect();
            }
        }
    }
}

@@ -860,24 +860,143 @@
 e.Splitter;%0A
+import com.google.common.util.concurrent.MoreExecutors;%0Aimport com.google.common.util.concurrent.ThreadFactoryBuilder;%0A
 import org.j
@@ -1033,16 +1033,16 @@
 tstrap;%0A
-
 import o
@@ -1380,55 +1380,9 @@
 ent.
-Executors;%0Aimport java.util.concurrent.TimeUnit
+*
 ;%0A%0Ai
@@ -1954,16 +1954,121 @@
 istener;
+%0A    private ExecutorService reconnectExecutor;%0A    private Graylog2Plugin.ReconnectRunnable reconnector;
 %0A%0A    pu
@@ -2877,25 +2877,326 @@
 ;%0A        %7D%0A
+        reconnectExecutor = Executors.newSingleThreadExecutor(%0A                new ThreadFactoryBuilder()%0A                        .setDaemon(true)%0A                        .setNameFormat(%22graylog2-appender-reconnect-%25d%22)%0A                        .build());%0A        reconnector = new ReconnectRunnable();
 %0A
-
     %7D%0A%0A    @
@@ -3728,166 +3728,45 @@
     
-new Thread(new Runnable() %7B%0A            @Override%0A            public void run() %7B%0A                channelFuture = reconnect();%0A            %7D%0A        %7D).start(
+reconnectExecutor.execute(reconnector
 );%0A 
@@ -4679,24 +4679,429 @@
 y();%0A    %7D%0A%0A
+    private class ReconnectRunnable implements Runnable %7B%0A        // TODO implement back off strategy here.%0A        // this implementation simply sleeps for now%0A        @Override%0A        public void run() %7B%0A            try %7B%0A                Thread.sleep(500);%0A            %7D catch (InterruptedException e) %7B%0A                // ignore%0A            %7D%0A            channelFuture = reconnect();%0A        %7D%0A    %7D%0A%0A
     private 
@@ -5518,136 +5518,34 @@
     
-// TODO implement back-off%0A                Thread.sleep(500);%0A                // could not connect, let's retry%0A                
+reconnectExecutor.execute(
 reco
@@ -5541,33 +5541,34 @@
 xecute(reconnect
-(
+or
 );%0A            %7D

tweak reconnection mechanism