/**
 *
 *  Copyright (c) 2017 Dell Inc., or its subsidiaries.
 *
 */
package com.emc.pravega.stream.impl.netty;


import static com.emc.pravega.common.netty.WireCommands.MAX_WIRECOMMAND_SIZE;

import java.security.NoSuchAlgorithmException;
import java.util.concurrent.CompletableFuture;

import javax.net.ssl.SSLException;

import com.emc.pravega.common.netty.AppendBatchSizeTracker;
import com.emc.pravega.common.netty.CommandDecoder;
import com.emc.pravega.common.netty.CommandEncoder;
import com.emc.pravega.common.netty.ExceptionLoggingHandler;
import com.emc.pravega.common.netty.PravegaNodeUri;
import com.emc.pravega.common.netty.ReplyProcessor;
import com.google.common.base.Preconditions;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.FingerprintTrustManagerFactory;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public final class ConnectionFactoryImpl implements ConnectionFactory {

    private final boolean ssl;
    private EventLoopGroup group;
    private boolean nio = false;

    /**
     * Actual implementation of ConnectionFactory interface.
     *
     * @param ssl Whether connection should use SSL or not.
     */
    public ConnectionFactoryImpl(boolean ssl) {
        this.ssl = ssl;
        try {
            this.group = new EpollEventLoopGroup();
        } catch (ExceptionInInitializerError | NoClassDefFoundError e) {
            log.warn("Epoll not available. Falling back on NIO.");
            nio = true;
            this.group = new NioEventLoopGroup();
        }
    }

    @Override
    public CompletableFuture<ClientConnection> establishConnection(PravegaNodeUri location, ReplyProcessor rp) {
        Preconditions.checkNotNull(location);
        final SslContext sslCtx;
        if (ssl) {
            try {
                sslCtx = SslContextBuilder.forClient()
                                          .trustManager(FingerprintTrustManagerFactory
                                                  .getInstance(FingerprintTrustManagerFactory.getDefaultAlgorithm()))
                                          .build();
            } catch (SSLException | NoSuchAlgorithmException e) {
                throw new RuntimeException(e);
            }
        } else {
            sslCtx = null;
        }
        AppendBatchSizeTracker batchSizeTracker = new AppendBatchSizeTrackerImpl();
        ClientConnectionInboundHandler handler = new ClientConnectionInboundHandler(location.getEndpoint(), rp, batchSizeTracker);
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(nio ? NioSocketChannel.class : EpollSocketChannel.class)
         .option(ChannelOption.TCP_NODELAY, true)
         .handler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
                 if (sslCtx != null) {
                     p.addLast(sslCtx.newHandler(ch.alloc(), location.getEndpoint(), location.getPort()));
                 }
                 // p.addLast(new LoggingHandler(LogLevel.INFO));
                 p.addLast(new ExceptionLoggingHandler(location.getEndpoint()),
                         new CommandEncoder(batchSizeTracker),
                         new LengthFieldBasedFrameDecoder(MAX_WIRECOMMAND_SIZE, 4, 4),
                         new CommandDecoder(),
                         handler);
             }
         });

        // Start the client.
        CompletableFuture<ClientConnection> result = new CompletableFuture<>();
        b.connect(location.getEndpoint(), location.getPort()).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (future.isSuccess()) {
                    result.complete(handler);
                } else {
                    result.completeExceptionally(future.cause());
                }
            }
        });
        return result;
    }

    @Override
    public void close() {
        // Shut down the event loop to terminate all threads.
        group.shutdownGracefully();
    }

    @Override
    protected void finalize() {
        group.shutdownGracefully();
    }
}
@@ -284,44 +284,134 @@
 re;%0A
-%0Aimport javax.net.ssl.SSLException;%0A
+import java.util.concurrent.atomic.AtomicBoolean;%0Aimport javax.net.ssl.SSLException;%0Aimport com.emc.pravega.common.Exceptions;
 %0Aimp
@@ -1740,16 +1740,83 @@
 = false;
+%0A    private final AtomicBoolean closed = new AtomicBoolean(false);
 %0A%0A    /*
@@ -2495,16 +2495,71 @@
 ation);%0A
+        Exceptions.checkNotClosed(closed.get(), this);%0A
         
@@ -4401,24 +4401,42 @@
 Future%3C%3E();%0A
+        try %7B%0A    
         b.co
@@ -4531,32 +4531,36 @@
 ) %7B%0A            
+    
 @Override%0A      
@@ -4545,32 +4545,36 @@
       @Override%0A
+    
             publ
@@ -4635,24 +4635,28 @@
             
+    
 if (future.i
@@ -4665,24 +4665,28 @@
 uccess()) %7B%0A
+    
             
@@ -4727,32 +4727,36 @@
                 
+    
 %7D else %7B%0A       
@@ -4760,32 +4760,36 @@
                 
+ 
+   
 result.completeE
@@ -4826,34 +4826,42 @@
                 
-%7D%0A
+    %7D%0A    
             %7D%0A  
@@ -4866,20 +4866,111 @@
         
+  
+  
 %7D);%0A
+        %7D catch (Exception e) %7B%0A            result.completeExceptionally(e);%0A        %7D%0A
         
@@ -5027,24 +5027,77 @@
 d close() %7B%0A
+        if (closed.compareAndSet(false, true)) %7B%0A    
         // S
@@ -5146,32 +5146,36 @@
 hreads.%0A        
+    
 group.shutdownGr
@@ -5182,24 +5182,34 @@
 acefully();%0A
+        %7D%0A
     %7D%0A%0A    @
@@ -5261,32 +5261,13 @@
     
-group.shutdownGracefully
+close
 ();%0A

Enforce connection factory close, and propagate exceptions in connection creation to the result. (#881)