package com.tacitknowledge.slowlight.proxyserver.server.proxy;

import com.tacitknowledge.slowlight.proxyserver.config.HandlerConfig;
import com.tacitknowledge.slowlight.proxyserver.handler.AbstractChannelHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/** @author Alexandr Donciu (adonciu@tacitknowledge.com) */
public class ProxyChannelHandler extends AbstractChannelHandler
{
    private static final Logger LOG = LoggerFactory.getLogger(ProxyChannelHandler.class);

    private String targetHost;
    private int targetPort;

    private Channel targetChannel;

    private EventLoopGroup clientWorkerGroup;

    public ProxyChannelHandler(final HandlerConfig handlerConfig, final String targetHost, final int targetPort,
        final EventLoopGroup clientWorkerGroup)
    {
        super(handlerConfig);

        this.targetHost = targetHost;
        this.targetPort = targetPort;
        this.clientWorkerGroup = clientWorkerGroup;
    }

    @Override
    public void channelActive(final ChannelHandlerContext ctx) throws Exception
    {
        targetChannel = connectTarget(ctx.channel());
    }

    @Override
    public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception
    {
        if (targetChannel.isActive())
        {
            targetChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess())
                    {
                        ctx.channel().read();
                    }
                    else
                    {
                        future.channel().close();
                    }
                }
            });
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception
    {
        closeOnFlush(targetChannel);
    }

    private Channel connectTarget(final Channel sourceChannel)
    {
        final Bootstrap clientBootstrap = new Bootstrap();
        clientBootstrap.group(clientWorkerGroup).channel(sourceChannel.getClass());
        clientBootstrap.option(ChannelOption.AUTO_READ, false);
        clientBootstrap.handler(new ProxyTargetChannelHandler(handlerConfig, sourceChannel));

        final ChannelFuture remoteChannelFuture = clientBootstrap.connect(new InetSocketAddress(targetHost, targetPort));

        remoteChannelFuture.addListener(new ChannelFutureListener()
        {
            @Override
            public void operationComplete(final ChannelFuture future) throws Exception
            {
                if (future.isSuccess())
                {
                    sourceChannel.read();
                }
                else
                {
                    sourceChannel.close();
                }
            }
        });

        return remoteChannelFuture.channel();
    }

    public String getTargetHost()
    {
        return targetHost;
    }

    public int getTargetPort()
    {
        return targetPort;
    }

    public EventLoopGroup getClientWorkerGroup()
    {
        return clientWorkerGroup;
    }
}

@@ -799,17 +799,16 @@
 class);%0A
-%0A
     priv
@@ -858,17 +858,16 @@
 etPort;%0A
-%0A
     priv
@@ -893,17 +893,16 @@
 hannel;%0A
-%0A
     priv
@@ -1053,16 +1053,39 @@
 etPort,%0A
+                       
         
@@ -1585,17 +1585,20 @@
     if (
-t
+getT
 argetCha
@@ -1601,16 +1601,18 @@
 tChannel
+()
 .isActiv
@@ -1642,52 +1642,35 @@
     
-target
+final 
 Channel
-.writeAndFlush(msg).add
+Future
 Listener
 (new
@@ -1657,38 +1657,34 @@
 elFutureListener
-(new C
+ c
 hannelFutureList
@@ -1691,410 +1691,135 @@
 ener
-() %7B%0A                @Override%0A                public void operationComplete(ChannelFuture future) throws Exception %7B%0A                    if (future.isSuccess())%0A                    %7B%0A                        ctx.channel().read();%0A                    %7D%0A                    else%0A                    %7B%0A                        future.channel().close();%0A                    %7D%0A                %7D%0A            %7D
+ = getChannelFutureListener(ctx.channel());%0A            getTargetChannel().writeAndFlush(msg).addListener(channelFutureListener
 );%0A 
@@ -1880,16 +1880,22 @@
 nactive(
+final 
 ChannelH
@@ -1957,17 +1957,20 @@
 OnFlush(
-t
+getT
 argetCha
@@ -1973,16 +1973,18 @@
 tChannel
+()
 );%0A    %7D
@@ -1987,29 +1987,31 @@
    %7D%0A%0A    pr
-ivate
+otected
  Channel con
@@ -2493,40 +2493,324 @@
     
-remoteChannelFuture.addListener(
+final ChannelFutureListener channelFutureListener = getChannelFutureListener(sourceChannel);%0A        remoteChannelFuture.addListener(channelFutureListener);%0A%0A        return remoteChannelFuture.channel();%0A    %7D%0A%0A    protected ChannelFutureListener getChannelFutureListener(final Channel channel)%0A    %7B%0A        return 
 new 
@@ -2907,22 +2907,16 @@
 omplete(
-final 
 ChannelF
@@ -3034,31 +3034,25 @@
             
-sourceC
+c
 hannel.read(
@@ -3131,23 +3131,17 @@
         
-sourceC
+c
 hannel.c
@@ -3193,56 +3193,8 @@
    %7D
-);%0A%0A        return remoteChannelFuture.channel()
 ;%0A  
@@ -3437,11 +3437,92 @@
 p;%0A    %7D
+%0A%0A    public Channel getTargetChannel()%0A    %7B%0A        return targetChannel;%0A    %7D
 %0A%7D%0A

refactor ProxyChannelHandler - extract future listener