/*
 * 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.master.transport;

import alluxio.grpc.MessagingServiceGrpc;
import alluxio.grpc.TransportMessage;
import alluxio.security.authentication.ClientContextServerInjector;

import com.google.common.base.MoreObjects;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;

/**
 * gRPC service handler for message exchange between messaging clients and servers
 * that are created by {@link GrpcMessagingTransport}.
 */
public class GrpcMessagingServiceClientHandler
    extends MessagingServiceGrpc.MessagingServiceImplBase {
  private static final Logger LOG =
      LoggerFactory.getLogger(GrpcMessagingServiceClientHandler.class);

  /** Messaging server provided listener for storing incoming connections. */
  private final Consumer<GrpcMessagingConnection> mListener;
  /** {@link GrpcMessagingServer}'s catalyst thread context. */
  private final GrpcMessagingContext mContext;
  /** Request timeout value for new connections. */
  private final long mRequestTimeoutMs;
  /** Executor for building server connections. */
  private final ExecutorService mExecutor;
  /** Address for the server that is serving this handler. */
  private final InetSocketAddress mServerAddress;

  /**
   * @param serverAddress server address
   * @param listener listener for incoming connections
   * @param context catalyst thread context
   * @param executor transport executor
   * @param requestTimeoutMs request timeout value for new connections
   */
  public GrpcMessagingServiceClientHandler(InetSocketAddress serverAddress,
      Consumer<GrpcMessagingConnection> listener, GrpcMessagingContext context,
      ExecutorService executor, long requestTimeoutMs) {
    mServerAddress = serverAddress;
    mListener = listener;
    mContext = context;
    mExecutor = executor;
    mRequestTimeoutMs = requestTimeoutMs;
  }

  /**
   * RPC for establishing bi-di stream between client/server.
   *
   * @param responseObserver client's stream observer
   * @return server's stream observer
   */
  @Override
  public StreamObserver<TransportMessage> connect(
      StreamObserver<TransportMessage> responseObserver) {
    // Transport level identifier for this connection.
    String transportId = MoreObjects.toStringHelper(this)
        .add("ServerAddress", mServerAddress)
        .add("ClientAddress", ClientContextServerInjector.getIpAddress())
        .toString();
    LOG.debug("Creating a messaging server connection: {}", transportId);

    // Create server connection that is bound to given client stream.
    GrpcMessagingConnection serverConnection =
        new GrpcMessagingServerConnection(transportId, mContext, mExecutor, mRequestTimeoutMs);
    serverConnection.setTargetObserver(responseObserver);
    LOG.debug("Created a messaging server connection: {}", serverConnection);

    // Update server with new connection.
    try {
      mContext.execute(() -> {
        mListener.accept(serverConnection);
      }).get();
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
      throw new RuntimeException(
          "Interrupted while waiting for server to register new connection.");
    } catch (ExecutionException ee) {
      throw new RuntimeException("Failed to register new connection with server", ee.getCause());
    }

    return serverConnection;
  }
}
