/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.network.connection;

import alluxio.Configuration;
import alluxio.resource.DynamicResourcePool;
import alluxio.util.ThreadFactoryUtils;

import com.google.common.base.Throwables;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;

import javax.annotation.PropertyKey;
import javax.annotation.concurrent.ThreadSafe;

/**
 * A pool to manage netty channels. Netty has it own implementation of channel pool but that
 * doesn't fit our needs for several reasons:
 * 1. We need a dynamic pool which can garbage collect idle channels.
 * 2. We need to have control on how channel is created. For example, our channel handler might
 *    not be annotated with {@code Sharable}. So we need to deep copy handlers when creating new
 *    channels.
 * 3. Netty channel pool interface is async which is not necessary for our usecase.
 */
@ThreadSafe
public final class NettyChannelPool extends DynamicResourcePool<Channel> {
  private static final int NETTY_CHANNEL_POOL_GC_THREADPOOL_SIZE = 10;
  private Callable<Bootstrap> mBootstrap;
  private final long mGcThresholdMs;
  private static final ScheduledExecutorService GC_EXECUTOR =
      new ScheduledThreadPoolExecutor(NETTY_CHANNEL_POOL_GC_THREADPOOL_SIZE,
          ThreadFactoryUtils.build("NettyChannelPoolGcThreads-%d", true));
  private static final boolean POOL_DISABLED =
      Configuration.getBoolean(alluxio.PropertyKey.USER_NETWORK_NETTY_CHANNEL_POOL_DISABLE);

  /**
   * Creates a netty channel pool instance with a minimum capacity of 1.
   *
   * @param bootstrap the netty bootstrap used to create netty channel
   * @param maxCapacity the maximum capacity of the pool
   * @param gcThresholdMs when a channel is older than this threshold and the pool's capacity
   *        is above the minimum capacity(1), it is closed and removed from the pool.
   */
  public NettyChannelPool(Callable<Bootstrap> bootstrap, int maxCapacity, long gcThresholdMs) {
    super(Options.defaultOptions().setMaxCapacity(maxCapacity).setGcExecutor(GC_EXECUTOR));
    mBootstrap = bootstrap;
    mGcThresholdMs = gcThresholdMs;
  }

  @Override
  protected void closeResource(Channel channel) {
    LOG.info("Channel closed");
    channel.close();
  }

  @Override
  protected void closeResourceSync(Channel channel) {
    LOG.info("Channel closed synchronously.");
    channel.close().syncUninterruptibly();
  }

  /**
   * Creates a netty channel instance.
   *
   * @return the channel created
   * @throws IOException if it fails to create a channel
   */
  @Override
  protected Channel createNewResource() throws IOException {
    Bootstrap bs;
    try {
      bs = mBootstrap.call();
    } catch (Exception e) {
      // No exception should happen here.
      throw Throwables.propagate(e);
    }
    try {
      ChannelFuture channelFuture = bs.connect().sync();
      if (channelFuture.isSuccess()) {
        LOG.info("Created netty channel to with netty boostrap {}.", mBootstrap);
        return channelFuture.channel();
      } else {
        LOG.error("Failed to create netty channel with netty boostrap {} and error {}.",
            mBootstrap, channelFuture.cause().getMessage());
        throw new IOException(channelFuture.cause());
      }
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * Checks whether a channel is healthy.
   *
   * @param channel the channel to check
   * @return true if the channel is active (i.e. connected)
   */
  @Override
  protected boolean isHealthy(Channel channel) {
    if (POOL_DISABLED) {
      // If we always returns false here, channels acquired by acquire will always be a newly
      // created channel. With this feature turned on, 1.3.0 client will be backward compatible
      // with 1.2.0 server.
      return false;
    }
    return channel.isActive();
  }

  @Override
  protected boolean shouldGc(ResourceInternal<Channel> channelResourceInternal) {
    return System.currentTimeMillis() - channelResourceInternal
        .getLastAccessTimeMs() > mGcThresholdMs;
  }
}

@@ -1747,87 +1747,8 @@
 10;%0A
-  private Callable%3CBootstrap%3E mBootstrap;%0A  private final long mGcThresholdMs;%0A
   pr
@@ -2096,16 +2096,95 @@
 ISABLE);
+%0A  private Callable%3CBootstrap%3E mBootstrap;%0A  private final long mGcThresholdMs;
 %0A%0A  /**%0A
@@ -4319,17 +4319,16 @@
 s return
-s
  false h
@@ -4353,23 +4353,42 @@
 ired by 
+NettyChannelPool#
 acquire
+()
  will al
@@ -4395,19 +4395,8 @@
 ways
- be a newly
 %0A   
@@ -4400,16 +4400,25 @@
       //
+ be newly
  created
@@ -4455,16 +4455,19 @@
 rned on,
+ %3E=
  1.3.0 c
@@ -4488,16 +4488,25 @@
 backward
+%0A      //
  compati
@@ -4512,22 +4512,16 @@
 ible
-%0A      //
  with
+ %3C=
  1.2

add doc