public boolean checkSend(EnumLite send, Class<?> sendClass, Class<?> receiveClass) {
  if (RpcConstants.EXTRA_DEBUGGING) {
    logger.debug(String.format("Checking send classes for send RpcType %s.  Send Class is %s and Receive class is %s.", send, sendClass, receiveClass));
  }
  RpcMessageType<?,?,?> type = sendMap.get(send);
  if (type == null) {
    throw new IllegalStateException(String.format("%s: There is no defined RpcMessage type for a Rpc send type of %s.", name, send));
  }

  if (type.getSend() != sendClass) {
    throw new IllegalStateException(String.format("%s: The definition for send doesn't match implementation code.  The definition is %s however the current send is trying to send an object of type %s.", name, type, sendClass.getCanonicalName()));
  }
  if (type.getRet() != receiveClass) {
    throw new IllegalStateException(String.format("%s: The definition for send doesn't match implementation code.  The definition is %s however the current send is trying to setup an expected reception of an object of type %s.", name, type, receiveClass.getCanonicalName()));
  }

  return true;
}
 

@Override
public <S extends ServerConnection<S>, T extends EnumLite>
void process(SaslResponseContext<S, T> context) throws Exception {
  final SaslMessage.Builder challenge = SaslMessage.newBuilder();
  final SaslServer saslServer = context.connection.getSaslServer();

  final byte[] challengeBytes = evaluateResponse(saslServer, context.saslResponse.getData().toByteArray());

  if (saslServer.isComplete()) {
    challenge.setStatus(SaslStatus.SASL_SUCCESS);
    if (challengeBytes != null) {
      challenge.setData(ByteString.copyFrom(challengeBytes));
    }

    handleSuccess(context, challenge, saslServer);
  } else {
    challenge.setStatus(SaslStatus.SASL_IN_PROGRESS)
        .setData(ByteString.copyFrom(challengeBytes));
    context.sender.send(new Response(context.saslResponseType, challenge.build()));
  }
}
 

/**
 * Creates various instances needed to start the SASL handshake. This is called from
 * {@link BasicClient#prepareSaslHandshake(RpcConnectionHandler, List)} only for
 * {@link org.apache.drill.exec.rpc.data.DataClient} and {@link org.apache.drill.exec.rpc.control.ControlClient}
 *
 * @param connectionHandler    - Connection handler used by client's to know about success/failure conditions.
 * @param serverAuthMechanisms - List of auth mechanisms configured on server side
 * @param connection - ClientConnection used for authentication
 * @param config - ClientConnection config
 * @param endpoint - Remote DrillbitEndpoint
 * @param client - Either of DataClient/ControlClient instance
 * @param saslRpcType - SASL_MESSAGE RpcType for Data and Control channel
 */
public static <T extends EnumLite, CC extends ClientConnection, HS extends MessageLite, HR extends MessageLite>
void prepareSaslHandshake(final RpcConnectionHandler<CC> connectionHandler, List<String> serverAuthMechanisms,
                          CC connection, BitConnectionConfig config, DrillbitEndpoint endpoint,
                          final BasicClient<T, CC, HS, HR> client, T saslRpcType) {
  try {
    final Map<String, String> saslProperties = SaslProperties.getSaslProperties(connection.isEncryptionEnabled(),
      connection.getMaxWrappedSize());
    final UserGroupInformation ugi = UserGroupInformation.getLoginUser();
    final AuthenticatorFactory factory = config.getAuthFactory(serverAuthMechanisms);
    client.startSaslHandshake(connectionHandler, config.getSaslClientProperties(endpoint, saslProperties),
      ugi, factory, saslRpcType);
  } catch (final IOException e) {
    logger.error("Failed while doing setup for starting sasl handshake for connection", connection.getName());
    final Exception ex = new RpcException(String.format("Failed to initiate authentication to %s",
      endpoint.getAddress()), e);
    connectionHandler.connectionFailed(RpcConnectionHandler.FailureType.AUTHENTICATION, ex);
  }
}
 

/**
 * Checks whether the oneof whose {@code oneofCase} has already been set. If so, an {@link
 * InvalidProtocolBufferException} is thrown.
 */
public static void throwIfOneofAlreadyWritten(
    JsonParser parser, Object oneofCase, String fieldName, boolean ignoreNull)
    throws InvalidProtocolBufferException {
  if (ignoreNull && parser.currentToken() == JsonToken.VALUE_NULL) {
    // If the value is null, we skip it and don't need to throw any error..
    return;
  }
  if (((EnumLite) oneofCase).getNumber() != 0) {
    // TODO: Add the actual variableName of the offending field to the error message like
    // upstream, not
    // too hard but just a little boring for the expected return.
    throw new InvalidProtocolBufferException(
        "Cannot set field "
            + fieldName
            + " because another field "
            + oneofCase
            + " belonging to the same oneof has already been set.");
  }
}
 

public boolean checkSend(EnumLite send, Class<?> sendClass, Class<?> receiveClass) {
  if (RpcConstants.EXTRA_DEBUGGING) {
    logger.debug(String.format("Checking send classes for send RpcType %s.  Send Class is %s and Receive class is %s.", send, sendClass, receiveClass));
  }
  RpcMessageType<?,?,?> type = sendMap.get(send);
  if (type == null) {
    throw new IllegalStateException(String.format("%s: There is no defined RpcMessage type for a Rpc send type of %s.", name, send));
  }

  if (type.getSend() != sendClass) {
    throw new IllegalStateException(String.format("%s: The definition for send doesn't match implementation code.  The definition is %s however the current send is trying to send an object of type %s.", name, type, sendClass.getCanonicalName()));
  }
  if (type.getRet() != receiveClass) {
    throw new IllegalStateException(String.format("%s: The definition for send doesn't match implementation code.  The definition is %s however the current send is trying to setup an expected reception of an object of type %s.", name, type, receiveClass.getCanonicalName()));
  }

  return true;
}
 

@SuppressWarnings("unchecked")
public SimpleProtocol(int protocolId, Map<Integer, ReceiveHandler<MessageLite, MessageLite>> handlers, BufferAllocator allocator, String name) {
  super();
  this.protocolId = protocolId;
  this.handlers = new ReceiveHandler[2048];
  this.defaultResponseInstances = new MessageLite[2048];
  this.defaultRequestInstances = new MessageLite[2048];
  RpcConfigBuilder builder = RpcConfig.newBuilder()
      .name(name)
      .timeout(0);
  for(Entry<Integer, ReceiveHandler<MessageLite, MessageLite>> e : handlers.entrySet()) {
    final int id = e.getKey();
    final ReceiveHandler<?,?> handler = e.getValue();
    final EnumLite num = new PseudoEnum(id);
    builder.add(num, (Class<? extends MessageLite>) handler.getDefaultRequest().getClass(), num, (Class<? extends MessageLite>) handler.getDefaultResponse().getClass());
    this.handlers[id] = e.getValue();
    this.defaultResponseInstances[id] = e.getValue().getDefaultResponse();
    this.defaultRequestInstances[id] = e.getValue().getDefaultRequest();
  }
  this.config = builder.build();
  this.allocator = allocator;
}
 

/** Get a hash code for given fields and values, using the given seed. */
@SuppressWarnings("unchecked")
protected static int hashFields(int hash, Map<FieldDescriptor, Object> map) {
  for (Map.Entry<FieldDescriptor, Object> entry : map.entrySet()) {
    FieldDescriptor field = entry.getKey();
    Object value = entry.getValue();
    hash = (37 * hash) + field.getNumber();
    if (field.isMapField()) {
      hash = (53 * hash) + hashMapField(value);
    } else if (field.getType() != FieldDescriptor.Type.ENUM){
      hash = (53 * hash) + value.hashCode();
    } else if (field.isRepeated()) {
      List<? extends EnumLite> list = (List<? extends EnumLite>) value;
      hash = (53 * hash) + Internal.hashEnumList(list);
    } else {
      hash = (53 * hash) + Internal.hashEnum((EnumLite) value);
    }
  }
  return hash;
}
 

/** Get a hash code for given fields and values, using the given seed. */
@SuppressWarnings("unchecked")
protected int hashFields(int hash, Map<FieldDescriptor, Object> map) {
  for (Map.Entry<FieldDescriptor, Object> entry : map.entrySet()) {
    FieldDescriptor field = entry.getKey();
    Object value = entry.getValue();
    hash = (37 * hash) + field.getNumber();
    if (field.getType() != FieldDescriptor.Type.ENUM){
      hash = (53 * hash) + value.hashCode();
    } else if (field.isRepeated()) {
      List<? extends EnumLite> list = (List<? extends EnumLite>) value;
      hash = (53 * hash) + hashEnumList(list);
    } else {
      hash = (53 * hash) + hashEnum((EnumLite) value);
    }
  }
  return hash;
}
 

/** Get a hash code for given fields and values, using the given seed. */
@SuppressWarnings("unchecked")
protected static int hashFields(int hash, Map<FieldDescriptor, Object> map) {
  for (Map.Entry<FieldDescriptor, Object> entry : map.entrySet()) {
    FieldDescriptor field = entry.getKey();
    Object value = entry.getValue();
    hash = (37 * hash) + field.getNumber();
    if (field.getType() != FieldDescriptor.Type.ENUM){
      hash = (53 * hash) + value.hashCode();
    } else if (field.isRepeated()) {
      List<? extends EnumLite> list = (List<? extends EnumLite>) value;
      hash = (53 * hash) + Internal.hashEnumList(list);
    } else {
      hash = (53 * hash) + Internal.hashEnum((EnumLite) value);
    }
  }
  return hash;
}
 

/** Get a hash code for given fields and values, using the given seed. */
@SuppressWarnings("unchecked")
protected int hashFields(int hash, Map<FieldDescriptor, Object> map) {
  for (Map.Entry<FieldDescriptor, Object> entry : map.entrySet()) {
    FieldDescriptor field = entry.getKey();
    Object value = entry.getValue();
    hash = (37 * hash) + field.getNumber();
    if (field.getType() != FieldDescriptor.Type.ENUM){
      hash = (53 * hash) + value.hashCode();
    } else if (field.isRepeated()) {
      List<? extends EnumLite> list = (List<? extends EnumLite>) value;
      hash = (53 * hash) + hashEnumList(list);
    } else {
      hash = (53 * hash) + hashEnum((EnumLite) value);
    }
  }
  return hash;
}
 

/** Get a hash code for given fields and values, using the given seed. */
@SuppressWarnings("unchecked")
protected int hashFields(int hash, Map<FieldDescriptor, Object> map) {
  for (Map.Entry<FieldDescriptor, Object> entry : map.entrySet()) {
    FieldDescriptor field = entry.getKey();
    Object value = entry.getValue();
    hash = (37 * hash) + field.getNumber();
    if (field.getType() != FieldDescriptor.Type.ENUM){
      hash = (53 * hash) + value.hashCode();
    } else if (field.isRepeated()) {
      List<? extends EnumLite> list = (List<? extends EnumLite>) value;
      hash = (53 * hash) + hashEnumList(list);
    } else {
      hash = (53 * hash) + hashEnum((EnumLite) value);
    }
  }
  return hash;
}
 

public <T extends MessageLite, E extends EnumLite, M extends MessageLite,
    R extends RpcCommand<T, C, E, M>> void runCommand(R cmd) {
//    if(logger.isDebugEnabled()) logger.debug(String.format("Running command %s sending to host %s:%d", cmd, host, port));
    C connection = connectionHolder.get();
    if (connection != null) {
      if (connection.isActive()) {
        cmd.connectionAvailable(connection);
//        logger.debug("Connection available and active, command run inline.");
        return;
      } else {
        // remove the old connection. (don't worry if we fail since someone else should have done it.
        connectionHolder.compareAndSet(connection, null);
      }
    }

    /**
     * We've arrived here without a connection, let's make sure only one of us makes a connection. (fyi, another
     * endpoint could create a reverse connection
     **/
    synchronized (this) {
      connection = connectionHolder.get();
      if (connection != null) {
        cmd.connectionAvailable(connection);

      } else {
//        logger.debug("No connection active, opening client connection.");
        BasicClient<?, C, HS, ?> client = getNewClient();
        ConnectionListeningFuture<T,E,M> future = new ConnectionListeningFuture<>(cmd);
        client.connectAsClient(future, handshake, host, port);
        future.waitAndRun();
//        logger.debug("Connection available and active, command now being run inline.");
      }
      return;

    }
  }
 

private RpcConfig(String name, Map<EnumLite, RpcMessageType<?, ?, ?>> sendMap,
    Map<Integer, RpcMessageType<?, ?, ?>> receiveMap, int timeout, Executor executor) {
  Preconditions.checkNotNull(executor, "Executor must be defined.");
  this.name = name;
  this.timeout = timeout;
  this.sendMap = ImmutableMap.copyOf(sendMap);
  this.receiveMap = ImmutableMap.copyOf(receiveMap);
  this.executor = executor;
}
 

public boolean checkResponseSend(EnumLite responseType, Class<?> responseClass) {
  if (RpcConstants.EXTRA_DEBUGGING) {
    logger.debug(String.format("Checking responce send of type %s with response class of %s.",  responseType, responseClass));
  }
  RpcMessageType<?,?,?> type = receiveMap.get(responseType.getNumber());
  if (type == null) {
    throw new IllegalStateException(String.format("%s: There is no defined RpcMessage type for a Rpc response of type %s.", name, responseType));
  }
  if (type.getRet() != responseClass) {
    throw new IllegalStateException(String.format("%s: The definition for the response doesn't match implementation code.  The definition is %s however the current response is trying to response with an object of type %s.", name, type, responseClass.getCanonicalName()));
  }

  return true;
}
 

@Override
public <S extends ServerConnection<S>, T extends EnumLite>
void process(SaslResponseContext<S, T> context) throws Exception {
  context.connection.initSaslServer(context.saslResponse.getMechanism());

  // assume #evaluateResponse must be called at least once
  RESPONSE_PROCESSORS.get(SaslStatus.SASL_IN_PROGRESS).process(context);
}
 

@Override
public <S extends ServerConnection<S>, T extends EnumLite>
void process(SaslResponseContext<S, T> context) throws Exception {
  final S connection = context.connection;
  logger.info("Client from {} failed authentication with encryption context:{} graciously, and does not want to " +
      "continue.", connection.getRemoteAddress().toString(), connection.getEncryptionCtxtString());
  throw new SaslException(String.format("Client graciously failed authentication. [Details: %s]",
      connection.getEncryptionCtxtString()));
}
 

private static <S extends ServerConnection<S>, T extends EnumLite>
void handleAuthFailure(final S connection, final ResponseSender sender,
                       final Exception e, final T saslResponseType) throws RpcException {
  final String remoteAddress = connection.getRemoteAddress().toString();

  logger.debug("Authentication using mechanism {} with encryption context {} failed from client {} due to {}",
      connection.getSaslServer().getMechanismName(), connection.getEncryptionCtxtString(), remoteAddress, e);

  // inform the client that authentication failed, and no more
  sender.send(new Response(saslResponseType, SASL_FAILED_MESSAGE));

  // drop connection
  throw new RpcException(e);
}
 

private RpcConfig(
    String name,
    Map<EnumLite, RpcMessageType<?, ?, ?>> sendMap,
    Map<Integer, RpcMessageType<?, ?, ?>> receiveMap,
    int timeout,
    Executor executor,
    Optional<SSLConfig> sslConfig) {
  this.name = name;
  this.timeout = timeout;
  this.sendMap = ImmutableMap.copyOf(sendMap);
  this.receiveMap = ImmutableMap.copyOf(receiveMap);
  this.executor = executor;
  this.sslConfig = sslConfig;
}
 

public boolean checkResponseSend(EnumLite responseType, Class<?> responseClass) {
  if (RpcConstants.EXTRA_DEBUGGING) {
    logger.debug(String.format("Checking responce send of type %s with response class of %s.",  responseType, responseClass));
  }
  RpcMessageType<?,?,?> type = receiveMap.get(responseType.getNumber());
  if (type == null) {
    throw new IllegalStateException(String.format("%s: There is no defined RpcMessage type for a Rpc response of type %s.", name, responseType));
  }
  if (type.getRet() != responseClass) {
    throw new IllegalStateException(String.format("%s: The definition for the response doesn't match implementation code.  The definition is %s however the current response is trying to response with an object of type %s.", name, type, responseClass.getCanonicalName()));
  }

  return true;
}
 

public <SEND extends MessageLite, RECEIVE extends MessageLite, T extends EnumLite>
RpcConfigBuilder add(T sendEnum, Class<SEND> send, T receiveEnum, Class<RECEIVE> rec) {
  RpcMessageType<SEND, RECEIVE, T> type = new RpcMessageType<SEND, RECEIVE, T>(sendEnum, send, receiveEnum, rec);
  this.sendMap.put(sendEnum, type);
  this.receiveMap.put(receiveEnum.getNumber(), type);
  return this;
}
 

public <SEND extends MessageLite, RECEIVE extends MessageLite> void send(
    RpcOutcomeListener<RECEIVE> outcomeListener,
    EnumLite rpcType,
    SEND protobufBody,
    Class<RECEIVE> clazz,
    ByteBuf... dataBodies) {
  assert rpcConfig.checkSend(rpcType, protobufBody.getClass(), clazz);
  connection.send(new ProxyListener<RECEIVE>(outcomeListener), RpcType.MESSAGE, msg(rpcType, protobufBody), FabricMessage.class, dataBodies);
}
 

public <SEND extends MessageLite, RECEIVE extends MessageLite> void sendUnsafe(
    RpcOutcomeListener<RECEIVE> outcomeListener,
    EnumLite rpcType,
    SEND protobufBody,
    Class<RECEIVE> clazz,
    ByteBuf... dataBodies) {
  assert rpcConfig.checkSend(rpcType, protobufBody.getClass(), clazz);
  connection.sendUnsafe(new ProxyListener<RECEIVE>(outcomeListener), RpcType.MESSAGE, msg(rpcType, protobufBody), FabricMessage.class, dataBodies);
}
 

private <SEND extends MessageLite> FabricMessage msg(EnumLite rpcType, SEND protobufBody){
  return FabricMessage.newBuilder()
      .setProtocolId(protocol.getProtocolId())
      .setInnerRpcType(rpcType.getNumber())
      .setMessage(protobufBody.toByteString())
      .build();
}
 

public EndpointCreator(FabricRunnerFactory factory, EnumLite num, Class<RESPONSE> responseClass, long timeout) {
  super();
  this.factory = factory;
  this.num = num;
  this.responseClass = responseClass;
  this.timeout = timeout;
}
 

protected PDFSCommand(Class<M> clazz, EnumLite rpcType, MessageLite request) {
  super();

  this.clazz = clazz;
  this.rpcType = rpcType;
  this.request = request;
}
 

private static int calculateHashCodeForObject(Object a) {
  if (a instanceof byte[]) {
    return Internal.hashCode((byte[]) a);
  }
  // Enums should be stored as integers internally.
  if (a instanceof EnumLite) {
    throw new UnsupportedOperationException();
  }
  return a.hashCode();
}
 

/**
 * @deprecated from v3.0.0-beta-3+, for compatibility with v2.5.0 and v2.6.1
 * generated code.
 */
@Deprecated
protected static int hashEnumList(List<? extends EnumLite> list) {
  int hash = 1;
  for (EnumLite e : list) {
    hash = 31 * hash + hashEnum(e);
  }
  return hash;
}
 

/** Helper method for implementing {@link Message#hashCode()}. */
protected static int hashEnumList(List<? extends EnumLite> list) {
  int hash = 1;
  for (EnumLite e : list) {
    hash = 31 * hash + hashEnum(e);
  }
  return hash;
}
 

/**
 * Calculate hash code for object.
 *
 * @param a the a
 * @return the int
 */
private static int calculateHashCodeForObject(Object a) {
    if (a instanceof byte[]) {
        return Internal.hashCode((byte[]) a);
    }
    // Enums should be stored as integers internally.
    if (a instanceof EnumLite) {
        throw new UnsupportedOperationException();
    }
    return a.hashCode();
}
 

/** Helper method for implementing {@link Message#hashCode()}. */
protected static int hashEnumList(List<? extends EnumLite> list) {
  int hash = 1;
  for (EnumLite e : list) {
    hash = 31 * hash + hashEnum(e);
  }
  return hash;
}