import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
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);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
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);
  // do not flush.  the context and first ipc call request must be sent
  // together to avoid possibility of broken pipes upon authz failure.
  // see writeConnectionHeader
  final ResponseBuffer buf = new ResponseBuffer();
  connectionContextHeader.writeDelimitedTo(buf);
  message.writeDelimitedTo(buf);
  synchronized (ipcStreams.out) {
    ipcStreams.sendRequest(buf.toByteArray());
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
/**
 * Process one RPC Request from buffer read from socket stream 
 *  - decode rpc in a rpc-Call
 *  - handle out-of-band RPC requests such as the initial connectionContext
 *  - A successfully decoded RpcCall will be deposited in RPC-Q and
 *    its response will be sent later when the request is processed.
 * 
 * Prior to this call the connectionHeader ("hrpc...") has been handled and
 * if SASL then SASL has been established and the buf we are passed
 * has been unwrapped from SASL.
 * 
 * @param buf - contains the RPC request header and the rpc request
 * @throws IOException - internal error that should not be returned to
 *         client, typically failure to respond to client
 * @throws WrappedRpcServerException - an exception that is sent back to the
 *         client in this method and does not require verbose logging by the
 *         Listener thread
 * @throws InterruptedException
 */    
private void processOneRpc(byte[] buf)
    throws IOException, WrappedRpcServerException, InterruptedException {
  int callId = -1;
  int retry = RpcConstants.INVALID_RETRY_COUNT;
  try {
    final DataInputStream dis =
        new DataInputStream(new ByteArrayInputStream(buf));
    final RpcRequestHeaderProto header =
        decodeProtobufFromStream(RpcRequestHeaderProto.newBuilder(), dis);
    callId = header.getCallId();
    retry = header.getRetryCount();
    if (LOG.isDebugEnabled()) {
      LOG.debug(" got #" + callId);
    }
    checkRpcHeaders(header);
    
    if (callId < 0) { // callIds typically used during connection setup
      processRpcOutOfBandRequest(header, dis);
    } else if (!connectionContextRead) {
      throw new WrappedRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "Connection context not established");
    } else {
      processRpcRequest(header, dis);
    }
  } catch (WrappedRpcServerException wrse) { // inform client of error
    Throwable ioe = wrse.getCause();
    final Call call = new Call(callId, retry, null, this);
    setupResponse(authFailedResponse, call,
        RpcStatusProto.FATAL, wrse.getRpcErrorCodeProto(), null,
        ioe.getClass().getName(), ioe.getMessage());
    call.sendResponse();
    throw wrse;
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
/**
 * Establish RPC connection setup by negotiating SASL if required, then
 * reading and authorizing the connection header
 * @param header - RPC header
 * @param dis - stream to request payload
 * @throws WrappedRpcServerException - setup failed due to SASL
 *         negotiation failure, premature or invalid connection context,
 *         or other state errors. This exception needs to be sent to the 
 *         client.
 * @throws IOException - failed to send a response back to the client
 * @throws InterruptedException
 */
private void processRpcOutOfBandRequest(RpcRequestHeaderProto header,
    DataInputStream dis) throws WrappedRpcServerException, IOException,
    InterruptedException {
  final int callId = header.getCallId();
  if (callId == CONNECTION_CONTEXT_CALL_ID) {
    // SASL must be established prior to connection context
    if (authProtocol == AuthProtocol.SASL && !saslContextEstablished) {
      throw new WrappedRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "Connection header sent during SASL negotiation");
    }
    // read and authorize the user
    processConnectionContext(dis);
  } else if (callId == AuthProtocol.SASL.callId) {
    // if client was switched to simple, ignore first SASL message
    if (authProtocol != AuthProtocol.SASL) {
      throw new WrappedRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "SASL protocol not requested by client");
    }
    saslReadAndProcess(dis);
  } else if (callId == PING_CALL_ID) {
    LOG.debug("Received ping message");
  } else {
    throw new WrappedRpcServerException(
        RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
        "Unknown out of band call #" + callId);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
@Test
public void testRpcClientId() {
  byte[] uuid = ClientId.getClientId();
  RpcRequestHeaderProto header = ProtoUtil.makeRpcRequestHeader(
      RpcKind.RPC_PROTOCOL_BUFFER, OperationProto.RPC_FINAL_PACKET, 0,
      RpcConstants.INVALID_RETRY_COUNT, uuid);
  assertTrue(Arrays.equals(uuid, header.getClientId().toByteArray()));
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
/**
 * Process an RPC Request - handle connection setup and decoding of
 * request into a Call
 * @param buf - contains the RPC request header and the rpc request
 * @throws IOException - internal error that should not be returned to
 *         client, typically failure to respond to client
 * @throws WrappedRpcServerException - an exception to be sent back to
 *         the client that does not require verbose logging by the
 *         Listener thread
 * @throws InterruptedException
 */    
private void processOneRpc(byte[] buf)
    throws IOException, WrappedRpcServerException, InterruptedException {
  int callId = -1;
  int retry = RpcConstants.INVALID_RETRY_COUNT;
  try {
    final DataInputStream dis =
        new DataInputStream(new ByteArrayInputStream(buf));
    final RpcRequestHeaderProto header =
        decodeProtobufFromStream(RpcRequestHeaderProto.newBuilder(), dis);
    callId = header.getCallId();
    retry = header.getRetryCount();
    if (LOG.isDebugEnabled()) {
      LOG.debug(" got #" + callId);
    }
    checkRpcHeaders(header);
    
    if (callId < 0) { // callIds typically used during connection setup
      processRpcOutOfBandRequest(header, dis);
    } else if (!connectionContextRead) {
      throw new WrappedRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "Connection context not established");
    } else {
      processRpcRequest(header, dis);
    }
  } catch (WrappedRpcServerException wrse) { // inform client of error
    Throwable ioe = wrse.getCause();
    final Call call = new Call(callId, retry, null, this);
    setupResponse(authFailedResponse, call,
        RpcStatusProto.FATAL, wrse.getRpcErrorCodeProto(), null,
        ioe.getClass().getName(), ioe.getMessage());
    responder.doRespond(call);
    throw wrse;
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
/**
 * Establish RPC connection setup by negotiating SASL if required, then
 * reading and authorizing the connection header
 * @param header - RPC header
 * @param dis - stream to request payload
 * @throws WrappedRpcServerException - setup failed due to SASL
 *         negotiation failure, premature or invalid connection context,
 *         or other state errors 
 * @throws IOException - failed to send a response back to the client
 * @throws InterruptedException
 */
private void processRpcOutOfBandRequest(RpcRequestHeaderProto header,
    DataInputStream dis) throws WrappedRpcServerException, IOException,
    InterruptedException {
  final int callId = header.getCallId();
  if (callId == CONNECTION_CONTEXT_CALL_ID) {
    // SASL must be established prior to connection context
    if (authProtocol == AuthProtocol.SASL && !saslContextEstablished) {
      throw new WrappedRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "Connection header sent during SASL negotiation");
    }
    // read and authorize the user
    processConnectionContext(dis);
  } else if (callId == AuthProtocol.SASL.callId) {
    // if client was switched to simple, ignore first SASL message
    if (authProtocol != AuthProtocol.SASL) {
      throw new WrappedRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "SASL protocol not requested by client");
    }
    saslReadAndProcess(dis);
  } else if (callId == PING_CALL_ID) {
    LOG.debug("Received ping message");
  } else {
    throw new WrappedRpcServerException(
        RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
        "Unknown out of band call #" + callId);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
/**
 * Establish RPC connection setup by negotiating SASL if required, then
 * reading and authorizing the connection header
 * @param header - RPC header
 * @param buffer - stream to request payload
 * @throws RpcServerException - setup failed due to SASL
 *         negotiation failure, premature or invalid connection context,
 *         or other state errors 
 * @throws IOException - failed to send a response back to the client
 * @throws InterruptedException
 */
private void processRpcOutOfBandRequest(RpcRequestHeaderProto header,
    RpcWritable.Buffer buffer) throws RpcServerException,
        IOException, InterruptedException {
  final int callId = header.getCallId();
  if (callId == CONNECTION_CONTEXT_CALL_ID) {
    // SASL must be established prior to connection context
    if (authProtocol == AuthProtocol.SASL && !saslContextEstablished) {
      throw new FatalRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "Connection header sent during SASL negotiation");
    }
    // read and authorize the user
    processConnectionContext(buffer);
  } else if (callId == AuthProtocol.SASL.callId) {
    // if client was switched to simple, ignore first SASL message
    if (authProtocol != AuthProtocol.SASL) {
      throw new FatalRpcServerException(
          RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
          "SASL protocol not requested by client");
    }
    saslReadAndProcess(buffer);
  } else if (callId == PING_CALL_ID) {
    LOG.debug("Received ping message");
  } else {
    throw new FatalRpcServerException(
        RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
        "Unknown out of band call #" + callId);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto; //导入依赖的package包/类
public RpcRequestMessageWrapper(
    RpcRequestHeaderProto requestHeader, Message theRequest) {
  super(requestHeader, theRequest);
}