/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.apache.drill.exec.rpc;

import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
import com.google.protobuf.Internal.EnumLite;
import com.google.protobuf.MessageLite;
import com.google.protobuf.Parser;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.proto.GeneralRPCProtos.RpcMode;
import org.apache.drill.exec.rpc.security.AuthenticationOutcomeListener;
import org.apache.drill.exec.rpc.security.AuthenticatorFactory;
import org.apache.hadoop.security.UserGroupInformation;

import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 *
 * @param <T> handshake rpc type
 * @param <CC> Client connection type
 * @param <HS> Handshake send type
 * @param <HR> Handshake receive type
 */
public abstract class BasicClient<T extends EnumLite, CC extends ClientConnection,
                                  HS extends MessageLite, HR extends MessageLite>
    extends RpcBus<T, CC> {
  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BasicClient.class);

  // The percentage of time that should pass before sending a ping message to ensure server doesn't time us out. For
  // example, if timeout is set to 30 seconds and we set percentage to 0.5, then if no write has happened within 15
  // seconds, the idle state handler will send a ping message.
  private static final double PERCENT_TIMEOUT_BEFORE_SENDING_PING = 0.5;

  private final Bootstrap b;
  protected CC connection;
  private final T handshakeType;
  private final Class<HR> responseClass;
  private final Parser<HR> handshakeParser;

  private final IdlePingHandler pingHandler;
  private ConnectionMultiListener.SSLHandshakeListener sslHandshakeListener = null;

  // Determines if authentication is completed between client and server
  private boolean authComplete = true;

  public BasicClient(RpcConfig rpcMapping, ByteBufAllocator alloc, EventLoopGroup eventLoopGroup, T handshakeType,
                     Class<HR> responseClass, Parser<HR> handshakeParser) {
    super(rpcMapping);
    this.responseClass = responseClass;
    this.handshakeType = handshakeType;
    this.handshakeParser = handshakeParser;
    final long timeoutInMillis = rpcMapping.hasTimeout() ?
        (long) (rpcMapping.getTimeout() * 1000.0 * PERCENT_TIMEOUT_BEFORE_SENDING_PING) :
        -1;
    this.pingHandler = rpcMapping.hasTimeout() ? new IdlePingHandler(timeoutInMillis) : null;

    b = new Bootstrap() //
        .group(eventLoopGroup) //
        .channel(TransportCheck.getClientSocketChannel()) //
        .option(ChannelOption.ALLOCATOR, alloc) //
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30 * 1000)
        .option(ChannelOption.SO_REUSEADDR, true)
        .option(ChannelOption.SO_RCVBUF, 1 << 17) //
        .option(ChannelOption.SO_SNDBUF, 1 << 17) //
        .option(ChannelOption.TCP_NODELAY, true)
        .handler(new ChannelInitializer<SocketChannel>() {

          @Override
          protected void initChannel(SocketChannel ch) throws Exception {
            // logger.debug("initializing client connection.");
            connection = initRemoteConnection(ch);

            ch.closeFuture().addListener(getCloseHandler(ch, connection));

            final ChannelPipeline pipe = ch.pipeline();
            // Make sure that the SSL handler is the first handler in the pipeline so everything is encrypted
            if (isSslEnabled()) {
              setupSSL(pipe, sslHandshakeListener);
            }

            pipe.addLast(RpcConstants.PROTOCOL_DECODER, getDecoder(connection.getAllocator()));
            pipe.addLast(RpcConstants.MESSAGE_DECODER, new RpcDecoder("c-" + rpcConfig.getName()));
            pipe.addLast(RpcConstants.PROTOCOL_ENCODER, new RpcEncoder("c-" + rpcConfig.getName()));
            pipe.addLast(RpcConstants.HANDSHAKE_HANDLER, new ClientHandshakeHandler(connection));

            if(pingHandler != null){
              pipe.addLast(RpcConstants.IDLE_STATE_HANDLER, pingHandler);
            }

            pipe.addLast(RpcConstants.MESSAGE_HANDLER, new InboundHandler(connection));
            pipe.addLast(RpcConstants.EXCEPTION_HANDLER, new RpcExceptionHandler<>(connection));
          }
        }); //

    // if(TransportCheck.SUPPORTS_EPOLL){
    // b.option(EpollChannelOption.SO_REUSEPORT, true); //
    // }
  }

  // Adds a SSL handler if enabled. Required only for client and server communications, so
  // a real implementation is only available for UserClient
  protected void setupSSL(ChannelPipeline pipe, ConnectionMultiListener.SSLHandshakeListener sslHandshakeListener) {
    throw new UnsupportedOperationException("SSL is implemented only by the User Client.");
  }

  protected boolean isSslEnabled() {
    return false;
  }

  /**
   * Set's the state for authentication complete.
   * @param authComplete - state to set. True means authentication between client and server is completed, false
   *                     means authentication is in progress.
   */
  protected void setAuthComplete(boolean authComplete) {
    this.authComplete = authComplete;
  }

  protected boolean isAuthComplete() {
    return authComplete;
  }

  // Save the SslChannel after the SSL handshake so it can be closed later
  public void setSslChannel(Channel c) {

  }

  @Override
  protected CC initRemoteConnection(SocketChannel channel){
    local=channel.localAddress();
    remote=channel.remoteAddress();
    return null;
  }

  private static final OutboundRpcMessage PING_MESSAGE = new OutboundRpcMessage(RpcMode.PING, 0, 0, Acks.OK);

  /**
   * Handler that watches for situations where we haven't read from the socket in a certain timeout. If we exceed this
   * timeout, we send a PING message to the server to state that we are still alive.
   */
  private class IdlePingHandler extends IdleStateHandler {

    private GenericFutureListener<Future<? super Void>> pingFailedHandler = new GenericFutureListener<Future<? super Void>>() {
      public void operationComplete(Future<? super Void> future) throws Exception {
        if (!future.isSuccess()) {
          logger.error("Unable to maintain connection {}.  Closing connection.", connection.getName());
          connection.close();
        }
      }
    };

    IdlePingHandler(long idleWaitInMillis) {
      super(0, idleWaitInMillis, 0, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
      if (evt.state() == IdleState.WRITER_IDLE) {
        ctx.writeAndFlush(PING_MESSAGE).addListener(pingFailedHandler);
      }
    }
  }

  public abstract ProtobufLengthDecoder getDecoder(BufferAllocator allocator);

  public boolean isActive() {
    return (connection != null) && connection.isActive();
  }

  protected abstract List<String> validateHandshake(HR validateHandshake) throws RpcException;

  /**
   * Creates various instances needed to start the SASL handshake. This is called from
   * {@link BasicClient#validateHandshake(MessageLite)} if authentication is required from server side.
   * @param connectionHandler - Connection handler used by client's to know about success/failure conditions.
   * @param serverAuthMechanisms - List of auth mechanisms configured on server side
   */
  protected abstract void prepareSaslHandshake(final RpcConnectionHandler<CC> connectionHandler,
                                               List<String> serverAuthMechanisms) throws RpcException;

  /**
   * Main method which starts the SASL handshake for all client channels (user/data/control) once it's determined
   * after regular RPC handshake that authentication is required by server side. Once authentication is completed
   * then only the underlying channel is made available to clients to send other RPC messages. Success and failure
   * events are notified to the connection handler on which client waits.
   * @param connectionHandler - Connection handler used by client's to know about success/failure conditions.
   * @param saslProperties - SASL related properties needed to create SASL client.
   * @param ugi - UserGroupInformation with logged in client side user
   * @param authFactory - Authentication factory to use for this SASL handshake.
   * @param rpcType - SASL_MESSAGE rpc type.
   */
  protected void startSaslHandshake(final RpcConnectionHandler<CC> connectionHandler,
                                    Map<String, ?> saslProperties, UserGroupInformation ugi,
                                    AuthenticatorFactory authFactory, T rpcType) {
    final String mechanismName = authFactory.getSimpleName();
    try {
      final SaslClient saslClient = authFactory.createSaslClient(ugi, saslProperties);
      if (saslClient == null) {
        final Exception ex = new SaslException(String.format("Cannot initiate authentication using %s mechanism. " +
          "Insufficient credentials or selected mechanism doesn't support configured security layers?", mechanismName));
        connectionHandler.connectionFailed(RpcConnectionHandler.FailureType.AUTHENTICATION, ex);
        return;
      }
      connection.setSaslClient(saslClient);
    } catch (final SaslException e) {
      logger.error("Failed while creating SASL client for SASL handshake for connection", connection.getName());
      connectionHandler.connectionFailed(RpcConnectionHandler.FailureType.AUTHENTICATION, e);
      return;
    }

    logger.debug("Initiating SASL exchange.");
    new AuthenticationOutcomeListener<>(this, connection, rpcType, ugi,
      new RpcOutcomeListener<Void>() {
      @Override
      public void failed(RpcException ex) {
        connectionHandler.connectionFailed(RpcConnectionHandler.FailureType.AUTHENTICATION, ex);
      }

      @Override
      public void success(Void value, ByteBuf buffer) {
        authComplete = true;
        connectionHandler.connectionSucceeded(connection);
      }

      @Override
      public void interrupted(InterruptedException ex) {
        connectionHandler.connectionFailed(RpcConnectionHandler.FailureType.AUTHENTICATION, ex);
      }
    }).initiate(mechanismName);
  }

  protected void finalizeConnection(HR handshake, CC connection) {
    // no-op
  }

  public <SEND extends MessageLite, RECEIVE extends MessageLite>
  void send(RpcOutcomeListener<RECEIVE> listener, T rpcType, SEND protobufBody,
            Class<RECEIVE> clazz, ByteBuf... dataBodies) {
    super.send(listener, connection, rpcType, protobufBody, clazz, dataBodies);
  }

  public <SEND extends MessageLite, RECEIVE extends MessageLite>
  DrillRpcFuture<RECEIVE> send(T rpcType, SEND protobufBody, Class<RECEIVE> clazz, ByteBuf... dataBodies) {
    return super.send(connection, rpcType, protobufBody, clazz, dataBodies);
  }

  public <SEND extends MessageLite, RECEIVE extends MessageLite>
  void send(RpcOutcomeListener<RECEIVE> listener, SEND protobufBody, boolean allowInEventLoop,
      ByteBuf... dataBodies) {
    super.send(listener, connection, handshakeType, protobufBody, (Class<RECEIVE>) responseClass,
        allowInEventLoop, dataBodies);
  }

  protected void connectAsClient(RpcConnectionHandler<CC> connectionListener, HS handshakeValue,
                                 String host, int port) {
    ConnectionMultiListener<T, CC, HS, HR, BasicClient<T, CC, HS, HR>> cml;
    ConnectionMultiListener.Builder<T, CC, HS, HR, BasicClient<T, CC, HS, HR> > builder =
        ConnectionMultiListener.newBuilder(connectionListener, handshakeValue, this);
    if (isSslEnabled()) {
      cml = builder.enableSSL().build();
      sslHandshakeListener = new ConnectionMultiListener.SSLHandshakeListener();
      sslHandshakeListener.setParent(cml);
    } else {
      cml = builder.build();
    }
    b.connect(host, port).addListener(cml.connectionHandler);
  }

  private class ClientHandshakeHandler extends AbstractHandshakeHandler<HR> {

    private final CC connection;

    ClientHandshakeHandler(CC connection) {
      super(BasicClient.this.handshakeType, BasicClient.this.handshakeParser);
      Preconditions.checkNotNull(connection);
      this.connection = connection;
    }

    @Override
    protected final void consumeHandshake(ChannelHandlerContext ctx, HR msg) throws Exception {
      // remove the handshake information from the queue so it doesn't sit there forever.
      final RpcOutcome<HR> response =
          connection.getAndRemoveRpcOutcome(handshakeType.getNumber(), coordinationId, responseClass);
      response.set(msg, null);
    }

  }

  public void setAutoRead(boolean enableAutoRead) {
    connection.setAutoRead(enableAutoRead);
  }

  public void close() {
    logger.debug("Closing client");

    if (connection != null) {
      connection.close();
      connection = null;
    }
  }
}
