package org.async.rmi.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import org.async.rmi.Connection;
import org.async.rmi.client.RemoteObjectAddress;
import org.async.rmi.messages.Message;
import org.async.rmi.messages.Response;
import org.async.rmi.pool.Pool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * Created by Barak Bar Orion
 * 27/10/14.
 */
public class NettyClientConnection implements Connection<Message> {
    private static final Logger logger = LoggerFactory.getLogger(NettyClientConnection.class);

    private final Bootstrap bootstrap;
    private Pool<Connection<Message>> pool;
    private RemoteObjectAddress address;
    private ChannelFuture channelFuture;
    private volatile boolean closed = false;
    private String localAddress;
    private String remoteAddress;

    public NettyClientConnection(Bootstrap bootstrap, RemoteObjectAddress address, Pool<Connection<Message>> pool) {
        this.bootstrap = bootstrap;
        this.address = address;
        this.pool = pool;
    }


    public CompletableFuture<Connection<Message>> connect() {
        final CompletableFuture<Connection<Message>> res = new CompletableFuture<>();

        channelFuture = bootstrap.connect(address.getHost(), address.getPort());
        channelFuture.addListener((ChannelFuture future) -> {
            try {
                Channel channel = future.sync().channel();
                InetSocketAddress la = (InetSocketAddress) channel.localAddress();
                localAddress = la.getHostString() + ":" + la.getPort();
                InetSocketAddress ra = (InetSocketAddress) channel.remoteAddress();
                remoteAddress = ra.getHostString() + ":" + ra.getPort();
                channel.closeFuture().addListener(cf -> {
                    closed = true;
                    pool.free(NettyClientConnection.this);
                });
                // To prevent premature message from the client to the server
                // resolve the connection future only when all handshakes are done.
                RMIClientHandler clientHandler = channel.pipeline().get(RMIClientHandler.class);
                clientHandler.getHandshakeCompleteFuture().whenComplete((aVoid, throwable) -> {
                    if (throwable != null) {
                        res.completeExceptionally(throwable);
                    } else {
                        res.complete(NettyClientConnection.this);
                    }
                });
            } catch (Exception e) {
                //noinspection ConstantConditions
                if (!(e instanceof java.net.ConnectException)) {
                    logger.error(e.toString(), e);
                }
            }
        });
        return res;
    }

    @Override
    public void free() {
        pool.free(this);
    }

    @Override
    public boolean isClosed() {
        return closed;
    }

    @Override
    public void close() throws IOException {
        closed = true;
        channelFuture.addListener((ChannelFuture future) -> future.channel().close());
    }

    @Override
    public void send(Message message) {
        channelFuture.addListener((ChannelFuture future) -> future.channel().writeAndFlush(message));
    }

    @Override
    public void sendOneWay(Message message, CompletableFuture<Response> response) {
        channelFuture.addListener((ChannelFuture future) -> future.channel().writeAndFlush(message)
                .addListener((ChannelFuture future1) -> {
                            try {
                                future1.get();
                                response.complete(null);
                            } catch (ExecutionException e) {
                                response.completeExceptionally(e.getCause());
                            } catch (Exception e) {
                                response.completeExceptionally(e);
                            }
                        }
                ));
    }

    public String getRemoteAddress() {
        return remoteAddress;
    }

    public String getLocalAddress() {
        return localAddress;
    }
}

@@ -161,24 +161,69 @@
 Connection;%0A
+import org.async.rmi.client.ClosedException;%0A
 import org.a
@@ -255,24 +255,24 @@
 ectAddress;%0A
-
 import org.a
@@ -1601,32 +1601,71 @@
           try %7B%0A
+                initAddresses(future);%0A
                 
@@ -1711,492 +1711,8 @@
 ();%0A
-                InetSocketAddress la = (InetSocketAddress) channel.localAddress();%0A                localAddress = la.getHostString() + %22:%22 + la.getPort();%0A                InetSocketAddress ra = (InetSocketAddress) channel.remoteAddress();%0A                remoteAddress = ra.getHostString() + %22:%22 + ra.getPort();%0A                channel.closeFuture().addListener(cf -%3E %7B%0A                    closed = true;%0A                    pool.free(NettyClientConnection.this);%0A                %7D);%0A
     
@@ -2298,32 +2298,275 @@
             %7D);%0A
+                channel.closeFuture().addListener(cf -%3E %7B%0A                    closed = true;%0A                    pool.free(NettyClientConnection.this);%0A                    res.completeExceptionally(new ClosedException());%0A                %7D);%0A%0A
             %7D ca
@@ -2787,32 +2787,32 @@
   %7D%0A        %7D);%0A
-
         return r
@@ -2814,32 +2814,503 @@
 urn res;%0A    %7D%0A%0A
+    private void initAddresses(ChannelFuture future)%7B%0A        try %7B%0A            Channel channel = future.sync().channel();%0A            InetSocketAddress la = (InetSocketAddress) channel.localAddress();%0A            localAddress = la.getHostString() + %22:%22 + la.getPort();%0A            InetSocketAddress ra = (InetSocketAddress) channel.remoteAddress();%0A            remoteAddress = ra.getHostString() + %22:%22 + ra.getPort();%0A        %7Dcatch(Exception ignored)%7B%0A        %7D%0A    %7D%0A%0A
     @Override%0A  

Handle server closing connection while client is in the handshake process.