private void writeConnectionContext(ConnectionId remoteId,
                                    AuthMethod authMethod)
                                        throws IOException {
  // Write out the ConnectionHeader
  IpcConnectionContextProto message = ProtoUtil.makeIpcConnectionContext(
      RPC.getProtocolName(remoteId.getProtocol()),
      remoteId.getTicket(),
      authMethod);
  RpcRequestHeaderProto connectionContextHeader = ProtoUtil
      .makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
          OperationProto.RPC_FINAL_PACKET, CONNECTION_CONTEXT_CALL_ID,
          RpcConstants.INVALID_RETRY_COUNT, clientId);
  RpcRequestMessageWrapper request =
      new RpcRequestMessageWrapper(connectionContextHeader, message);
  
  // Write out the packet length
  out.writeInt(request.getLength());
  request.write(out);
}
 

private void writeConnectionContext(ConnectionId remoteId,
                                    AuthMethod authMethod)
                                        throws IOException {
  // Write out the ConnectionHeader
  IpcConnectionContextProto message = ProtoUtil.makeIpcConnectionContext(
      RPC.getProtocolName(remoteId.getProtocol()),
      remoteId.getTicket(),
      authMethod);
  RpcRequestHeaderProto connectionContextHeader = ProtoUtil
      .makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
          OperationProto.RPC_FINAL_PACKET, CONNECTION_CONTEXT_CALL_ID,
          RpcConstants.INVALID_RETRY_COUNT, clientId);
  RpcRequestMessageWrapper request =
      new RpcRequestMessageWrapper(connectionContextHeader, message);
  
  // Write out the packet length
  out.writeInt(request.getLength());
  request.write(out);
}
 

public Class<? extends Writable> getRpcRequestWrapper(
    RpcKindProto rpcKind) {
  if (rpcRequestClass != null)
     return rpcRequestClass;
  RpcKindMapValue val = rpcKindMap.get(ProtoUtil.convert(rpcKind));
  return (val == null) ? null : val.rpcRequestWrapperClass; 
}
 

public Class<? extends Writable> getRpcRequestWrapper(
    RpcKindProto rpcKind) {
  if (rpcRequestClass != null)
     return rpcRequestClass;
  RpcKindMapValue val = rpcKindMap.get(ProtoUtil.convert(rpcKind));
  return (val == null) ? null : val.rpcRequestWrapperClass; 
}
 

/** Reads the connection context following the connection header
 * @param dis - DataInputStream from which to read the header 
 * @throws WrappedRpcServerException - if the header cannot be
 *         deserialized, or the user is not authorized
 */ 
private void processConnectionContext(DataInputStream dis)
    throws WrappedRpcServerException {
  // allow only one connection context during a session
  if (connectionContextRead) {
    throw new WrappedRpcServerException(
        RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
        "Connection context already processed");
  }
  connectionContext = decodeProtobufFromStream(
      IpcConnectionContextProto.newBuilder(), dis);
  protocolName = connectionContext.hasProtocol() ? connectionContext
      .getProtocol() : null;

  UserGroupInformation protocolUser = ProtoUtil.getUgi(connectionContext);
  if (saslServer == null) {
    user = protocolUser;
  } else {
    // user is authenticated
    user.setAuthenticationMethod(authMethod);
    //Now we check if this is a proxy user case. If the protocol user is
    //different from the 'user', it is a proxy user scenario. However, 
    //this is not allowed if user authenticated with DIGEST.
    if ((protocolUser != null)
        && (!protocolUser.getUserName().equals(user.getUserName()))) {
      if (authMethod == AuthMethod.TOKEN) {
        // Not allowed to doAs if token authentication is used
        throw new WrappedRpcServerException(
            RpcErrorCodeProto.FATAL_UNAUTHORIZED,
            new AccessControlException("Authenticated user (" + user
                + ") doesn't match what the client claims to be ("
                + protocolUser + ")"));
      } else {
        // Effective user can be different from authenticated user
        // for simple auth or kerberos auth
        // The user is the real user. Now we create a proxy user
        UserGroupInformation realUser = user;
        user = UserGroupInformation.createProxyUser(protocolUser
            .getUserName(), realUser);
      }
    }
  }
  authorizeConnection();
  // don't set until after authz because connection isn't established
  connectionContextRead = true;
}
 

private static byte[] readVarintBytes(DataInput in) throws IOException {
  final int length = ProtoUtil.readRawVarint32(in);
  final byte[] bytes = new byte[length];
  in.readFully(bytes);
  return bytes;
}
 

@Override
public void readFields(DataInput in) throws IOException {
  int length = ProtoUtil.readRawVarint32(in);
  theResponseRead = new byte[length];
  in.readFully(theResponseRead);
}
 

public Connection(ConnectionId remoteId, int serviceClass) throws IOException {
  this.remoteId = remoteId;
  this.server = remoteId.getAddress();
  if (server.isUnresolved()) {
    throw NetUtils.wrapException(server.getHostName(),
        server.getPort(),
        null,
        0,
        new UnknownHostException());
  }
  this.rpcTimeout = remoteId.getRpcTimeout();
  this.maxIdleTime = remoteId.getMaxIdleTime();
  this.connectionRetryPolicy = remoteId.connectionRetryPolicy;
  this.maxRetriesOnSasl = remoteId.getMaxRetriesOnSasl();
  this.maxRetriesOnSocketTimeouts = remoteId.getMaxRetriesOnSocketTimeouts();
  this.tcpNoDelay = remoteId.getTcpNoDelay();
  this.doPing = remoteId.getDoPing();
  if (doPing) {
    // construct a RPC header with the callId as the ping callId
    pingRequest = new ByteArrayOutputStream();
    RpcRequestHeaderProto pingHeader = ProtoUtil
        .makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
            OperationProto.RPC_FINAL_PACKET, PING_CALL_ID,
            RpcConstants.INVALID_RETRY_COUNT, clientId);
    pingHeader.writeDelimitedTo(pingRequest);
  }
  this.pingInterval = remoteId.getPingInterval();
  this.serviceClass = serviceClass;
  if (LOG.isDebugEnabled()) {
    LOG.debug("The ping interval is " + this.pingInterval + " ms.");
  }

  UserGroupInformation ticket = remoteId.getTicket();
  // try SASL if security is enabled or if the ugi contains tokens.
  // this causes a SIMPLE client with tokens to attempt SASL
  boolean trySasl = UserGroupInformation.isSecurityEnabled() ||
                    (ticket != null && !ticket.getTokens().isEmpty());
  this.authProtocol = trySasl ? AuthProtocol.SASL : AuthProtocol.NONE;
  
  this.setName("IPC Client (" + socketFactory.hashCode() +") connection to " +
      server.toString() +
      " from " + ((ticket==null)?"an unknown user":ticket.getUserName()));
  this.setDaemon(true);
}
 

/** Reads the connection context following the connection header
 * @param dis - DataInputStream from which to read the header 
 * @throws WrappedRpcServerException - if the header cannot be
 *         deserialized, or the user is not authorized
 */ 
private void processConnectionContext(DataInputStream dis)
    throws WrappedRpcServerException {
  // allow only one connection context during a session
  if (connectionContextRead) {
    throw new WrappedRpcServerException(
        RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
        "Connection context already processed");
  }
  connectionContext = decodeProtobufFromStream(
      IpcConnectionContextProto.newBuilder(), dis);
  protocolName = connectionContext.hasProtocol() ? connectionContext
      .getProtocol() : null;

  UserGroupInformation protocolUser = ProtoUtil.getUgi(connectionContext);
  if (saslServer == null) {
    user = protocolUser;
  } else {
    // user is authenticated
    user.setAuthenticationMethod(authMethod);
    //Now we check if this is a proxy user case. If the protocol user is
    //different from the 'user', it is a proxy user scenario. However, 
    //this is not allowed if user authenticated with DIGEST.
    if ((protocolUser != null)
        && (!protocolUser.getUserName().equals(user.getUserName()))) {
      if (authMethod == AuthMethod.TOKEN) {
        // Not allowed to doAs if token authentication is used
        throw new WrappedRpcServerException(
            RpcErrorCodeProto.FATAL_UNAUTHORIZED,
            new AccessControlException("Authenticated user (" + user
                + ") doesn't match what the client claims to be ("
                + protocolUser + ")"));
      } else {
        // Effective user can be different from authenticated user
        // for simple auth or kerberos auth
        // The user is the real user. Now we create a proxy user
        UserGroupInformation realUser = user;
        user = UserGroupInformation.createProxyUser(protocolUser
            .getUserName(), realUser);
      }
    }
  }
  authorizeConnection();
  // don't set until after authz because connection isn't established
  connectionContextRead = true;
}
 

private static byte[] readVarintBytes(DataInput in) throws IOException {
  final int length = ProtoUtil.readRawVarint32(in);
  final byte[] bytes = new byte[length];
  in.readFully(bytes);
  return bytes;
}
 

@Override
public void readFields(DataInput in) throws IOException {
  int length = ProtoUtil.readRawVarint32(in);
  theResponseRead = new byte[length];
  in.readFully(theResponseRead);
}
 

public Connection(ConnectionId remoteId, int serviceClass) throws IOException {
  this.remoteId = remoteId;
  this.server = remoteId.getAddress();
  if (server.isUnresolved()) {
    throw NetUtils.wrapException(server.getHostName(),
        server.getPort(),
        null,
        0,
        new UnknownHostException());
  }
  this.rpcTimeout = remoteId.getRpcTimeout();
  this.maxIdleTime = remoteId.getMaxIdleTime();
  this.connectionRetryPolicy = remoteId.connectionRetryPolicy;
  this.maxRetriesOnSasl = remoteId.getMaxRetriesOnSasl();
  this.maxRetriesOnSocketTimeouts = remoteId.getMaxRetriesOnSocketTimeouts();
  this.tcpNoDelay = remoteId.getTcpNoDelay();
  this.doPing = remoteId.getDoPing();
  if (doPing) {
    // construct a RPC header with the callId as the ping callId
    pingRequest = new ByteArrayOutputStream();
    RpcRequestHeaderProto pingHeader = ProtoUtil
        .makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
            OperationProto.RPC_FINAL_PACKET, PING_CALL_ID,
            RpcConstants.INVALID_RETRY_COUNT, clientId);
    pingHeader.writeDelimitedTo(pingRequest);
  }
  this.pingInterval = remoteId.getPingInterval();
  this.serviceClass = serviceClass;
  if (LOG.isDebugEnabled()) {
    LOG.debug("The ping interval is " + this.pingInterval + " ms.");
  }

  UserGroupInformation ticket = remoteId.getTicket();
  // try SASL if security is enabled or if the ugi contains tokens.
  // this causes a SIMPLE client with tokens to attempt SASL
  boolean trySasl = UserGroupInformation.isSecurityEnabled() ||
                    (ticket != null && !ticket.getTokens().isEmpty());
  this.authProtocol = trySasl ? AuthProtocol.SASL : AuthProtocol.NONE;
  
  this.setName("IPC Client (" + socketFactory.hashCode() +") connection to " +
      server.toString() +
      " from " + ((ticket==null)?"an unknown user":ticket.getUserName()));
  this.setDaemon(true);
}
 

private static byte[] readVarintBytes(DataInput in) throws IOException {
  final int length = ProtoUtil.readRawVarint32(in);
  final byte[] bytes = new byte[length];
  in.readFully(bytes);
  return bytes;
}
 

@Override
public void readFields(DataInput in) throws IOException {
  int length = ProtoUtil.readRawVarint32(in);
  theResponseRead = new byte[length];
  in.readFully(theResponseRead);
}