/*
 * Copyright (c) "Neo4j"
 * Neo4j Sweden AB [http://neo4j.com]
 *
 * This file is part of Neo4j.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.neo4j.driver.internal.async.pool;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.pool.ChannelHealthChecker;
import io.netty.channel.pool.FixedChannelPool;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicBoolean;

import org.neo4j.driver.internal.BoltServerAddress;
import org.neo4j.driver.internal.async.connection.ChannelConnector;
import org.neo4j.driver.internal.metrics.ListenerEvent;

import static java.util.Objects.requireNonNull;
import static org.neo4j.driver.internal.async.connection.ChannelAttributes.setPoolId;
import static org.neo4j.driver.internal.util.Futures.asCompletionStage;

public class NettyChannelPool implements ExtendedChannelPool
{
    /**
     * Unlimited amount of parties are allowed to request channels from the pool.
     */
    private static final int MAX_PENDING_ACQUIRES = Integer.MAX_VALUE;
    /**
     * Do not check channels when they are returned to the pool.
     */
    private static final boolean RELEASE_HEALTH_CHECK = false;

    private final FixedChannelPool delegate;
    private final AtomicBoolean closed = new AtomicBoolean( false );
    private final String id;
    private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();

    NettyChannelPool( BoltServerAddress address, ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker handler,
            ChannelHealthChecker healthCheck, long acquireTimeoutMillis, int maxConnections )
    {
        requireNonNull( address );
        requireNonNull( connector );
        requireNonNull( handler );
        this.id = poolId( address );
        this.delegate = new FixedChannelPool( bootstrap, handler, healthCheck, FixedChannelPool.AcquireTimeoutAction.FAIL, acquireTimeoutMillis, maxConnections,
                MAX_PENDING_ACQUIRES, RELEASE_HEALTH_CHECK )
        {
            @Override
            protected ChannelFuture connectChannel( Bootstrap bootstrap )
            {
                ListenerEvent creatingEvent = handler.channelCreating( id );
                ChannelFuture channelFuture = connector.connect( address, bootstrap );
                channelFuture.addListener( future -> {
                    if ( future.isSuccess() )
                    {
                        // notify pool handler about a successful connection
                        Channel channel = channelFuture.channel();
                        setPoolId( channel, id );
                        handler.channelCreated( channel, creatingEvent );
                    }
                    else
                    {
                        handler.channelFailedToCreate( id );
                    }
                } );
                return channelFuture;
            }
        };
    }

    @Override
    public CompletionStage<Void> close()
    {
        if ( closed.compareAndSet( false, true ) )
        {
            asCompletionStage( delegate.closeAsync(), closeFuture );
        }
        return closeFuture;
    }

    @Override
    public CompletionStage<Channel> acquire()
    {
        return asCompletionStage( delegate.acquire() );
    }

    @Override
    public CompletionStage<Void> release( Channel channel )
    {
        return asCompletionStage( delegate.release( channel ) );
    }

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

    @Override
    public String id()
    {
        return this.id;
    }

    private String poolId( BoltServerAddress serverAddress )
    {
        return String.format( "%s:%d-%d", serverAddress.host(), serverAddress.port(), this.hashCode() );
    }
}

@@ -807,24 +807,64 @@
 nnelFuture;%0A
+import io.netty.channel.ChannelPromise;%0A
 import io.ne
@@ -2902,16 +2902,25 @@
 Future c
+onnectedC
 hannelFu
@@ -2987,47 +2987,381 @@
     
-channelFuture.addListener( future -%3E %7B%0A
+Channel channel = connectedChannelFuture.channel();%0A                // This ensures that handler.channelCreated is called before SimpleChannelPool calls handler.channelAcquired%0A                ChannelPromise trackedChannelFuture = channel.newPromise();%0A                connectedChannelFuture.addListener(%0A                        future -%3E%0A                        %7B%0A        
     
@@ -3414,34 +3414,47 @@
                 
-%7B%0A
+        %7B%0A     
                 
@@ -3461,16 +3461,19 @@
         
+   
 // notif
@@ -3529,67 +3529,8 @@
     
-                Channel channel = channelFuture.channel();%0A
     
@@ -3571,24 +3571,32 @@
 nnel, id );%0A
+        
             
@@ -3669,34 +3669,117 @@
                 
-%7D%0A
+            trackedChannelFuture.setSuccess();%0A                            %7D%0A        
                 
@@ -3799,34 +3799,50 @@
                 
+     
-%7B%0A
+   %7B%0A        
                 
@@ -3898,34 +3898,133 @@
                 
-%7D%0A
+            trackedChannelFuture.setFailure( future.cause() );%0A                            %7D%0A        
                 
@@ -4051,17 +4051,24 @@
  return 
-c
+trackedC
 hannelFu

Fix NettyChannelTracker race condition when tracking new channels (#904)