import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private void wrapWithSasl(RpcCall call) throws IOException {
  if (call.connection.saslServer != null) {
    byte[] token = call.rpcResponse.array();
    // synchronization may be needed since there can be multiple Handler
    // threads using saslServer to wrap responses.
    synchronized (call.connection.saslServer) {
      token = call.connection.saslServer.wrap(token, 0, token.length);
    }
    if (LOG.isDebugEnabled())
      LOG.debug("Adding saslServer wrapped token of size " + token.length
          + " as call response.");
    // rebuild with sasl header and payload
    RpcResponseHeaderProto saslHeader = RpcResponseHeaderProto.newBuilder()
        .setCallId(AuthProtocol.SASL.callId)
        .setStatus(RpcStatusProto.SUCCESS)
        .build();
    RpcSaslProto saslMessage = RpcSaslProto.newBuilder()
        .setState(SaslState.WRAP)
        .setToken(ByteString.copyFrom(token))
        .build();
    setupResponse(call, saslHeader, RpcWritable.wrap(saslMessage));
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
@InterfaceStability.Unstable
@InterfaceAudience.LimitedPrivate({"HDFS"})
public void abortResponse(Throwable t) throws IOException {
  // don't send response if the call was already sent or aborted.
  if (responseWaitCount.getAndSet(-1) > 0) {
    // clone the call to prevent a race with the other thread stomping
    // on the response while being sent.  the original call is
    // effectively discarded since the wait count won't hit zero
    Call call = new Call(this);
    setupResponse(new ByteArrayOutputStream(), call,
        RpcStatusProto.FATAL, RpcErrorCodeProto.ERROR_RPC_SERVER,
        null, t.getClass().getName(), StringUtils.stringifyException(t));
    call.sendResponse();
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private void doSaslReply(Message message) throws IOException {
  final Call saslCall = new Call(AuthProtocol.SASL.callId,
      RpcConstants.INVALID_RETRY_COUNT, null, this);
  final ByteArrayOutputStream saslResponse = new ByteArrayOutputStream();
  setupResponse(saslResponse, saslCall,
      RpcStatusProto.SUCCESS, null,
      new RpcResponseWrapper(message), null, null);
  saslCall.sendResponse();
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的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.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private static void wrapWithSasl(ByteArrayOutputStream response, Call call)
    throws IOException {
  if (call.connection.saslServer != null) {
    byte[] token = call.rpcResponse.array();
    // synchronization may be needed since there can be multiple Handler
    // threads using saslServer to wrap responses.
    synchronized (call.connection.saslServer) {
      token = call.connection.saslServer.wrap(token, 0, token.length);
    }
    if (LOG.isDebugEnabled())
      LOG.debug("Adding saslServer wrapped token of size " + token.length
          + " as call response.");
    response.reset();
    // rebuild with sasl header and payload
    RpcResponseHeaderProto saslHeader = RpcResponseHeaderProto.newBuilder()
        .setCallId(AuthProtocol.SASL.callId)
        .setStatus(RpcStatusProto.SUCCESS)
        .build();
    RpcSaslProto saslMessage = RpcSaslProto.newBuilder()
        .setState(SaslState.WRAP)
        .setToken(ByteString.copyFrom(token, 0, token.length))
        .build();
    RpcResponseMessageWrapper saslResponse =
        new RpcResponseMessageWrapper(saslHeader, saslMessage);

    DataOutputStream out = new DataOutputStream(response);
    out.writeInt(saslResponse.getLength());
    saslResponse.write(out);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private void doSaslReply(Message message) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Sending sasl message "+message);
  }
  setupResponse(saslResponse, saslCall,
      RpcStatusProto.SUCCESS, null,
      new RpcResponseWrapper(message), null, null);
  responder.doRespond(saslCall);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的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.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private void wrapWithSasl(ByteArrayOutputStream response, Call call)
    throws IOException {
  if (call.connection.saslServer != null) {
    byte[] token = response.toByteArray();
    // synchronization may be needed since there can be multiple Handler
    // threads using saslServer to wrap responses.
    synchronized (call.connection.saslServer) {
      token = call.connection.saslServer.wrap(token, 0, token.length);
    }
    if (LOG.isDebugEnabled())
      LOG.debug("Adding saslServer wrapped token of size " + token.length
          + " as call response.");
    response.reset();
    // rebuild with sasl header and payload
    RpcResponseHeaderProto saslHeader = RpcResponseHeaderProto.newBuilder()
        .setCallId(AuthProtocol.SASL.callId)
        .setStatus(RpcStatusProto.SUCCESS)
        .build();
    RpcSaslProto saslMessage = RpcSaslProto.newBuilder()
        .setState(SaslState.WRAP)
        .setToken(ByteString.copyFrom(token, 0, token.length))
        .build();
    RpcResponseMessageWrapper saslResponse =
        new RpcResponseMessageWrapper(saslHeader, saslMessage);

    DataOutputStream out = new DataOutputStream(response);
    out.writeInt(saslResponse.getLength());
    saslResponse.write(out);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
@Override
void doResponse(Throwable t) throws IOException {
  RpcCall call = this;
  if (t != null) {
    // clone the call to prevent a race with another thread stomping
    // on the response while being sent.  the original call is
    // effectively discarded since the wait count won't hit zero
    call = new RpcCall(this);
    setupResponse(call,
        RpcStatusProto.FATAL, RpcErrorCodeProto.ERROR_RPC_SERVER,
        null, t.getClass().getName(), StringUtils.stringifyException(t));
  }
  connection.sendResponse(call);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private void doSaslReply(Message message) throws IOException {
  final RpcCall saslCall = new RpcCall(this, AuthProtocol.SASL.callId);
  setupResponse(saslCall,
      RpcStatusProto.SUCCESS, null,
      RpcWritable.wrap(message), null, null);
  sendResponse(saslCall);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
/**
 * Setup response for the IPC Call.
 * 
 * @param responseBuf buffer to serialize the response into
 * @param call {@link Call} to which we are setting up the response
 * @param status of the IPC call
 * @param rv return value for the IPC Call, if the call was successful
 * @param errorClass error class, if the the call failed
 * @param error error message, if the call failed
 * @throws IOException
 */
private void setupResponse(
    RpcCall call, RpcStatusProto status, RpcErrorCodeProto erCode,
    Writable rv, String errorClass, String error)
        throws IOException {
  // fatal responses will cause the reader to close the connection.
  if (status == RpcStatusProto.FATAL) {
    call.connection.setShouldClose();
  }
  RpcResponseHeaderProto.Builder headerBuilder =
      RpcResponseHeaderProto.newBuilder();
  headerBuilder.setClientId(ByteString.copyFrom(call.clientId));
  headerBuilder.setCallId(call.callId);
  headerBuilder.setRetryCount(call.retryCount);
  headerBuilder.setStatus(status);
  headerBuilder.setServerIpcVersionNum(CURRENT_VERSION);

  if (status == RpcStatusProto.SUCCESS) {
    RpcResponseHeaderProto header = headerBuilder.build();
    try {
      setupResponse(call, header, rv);
    } catch (Throwable t) {
      LOG.warn("Error serializing call response for call " + call, t);
      // Call back to same function - this is OK since the
      // buffer is reset at the top, and since status is changed
      // to ERROR it won't infinite loop.
      setupResponse(call, RpcStatusProto.ERROR,
          RpcErrorCodeProto.ERROR_SERIALIZING_RESPONSE,
          null, t.getClass().getName(),
          StringUtils.stringifyException(t));
      return;
    }
  } else { // Rpc Failure
    headerBuilder.setExceptionClassName(errorClass);
    headerBuilder.setErrorMsg(error);
    headerBuilder.setErrorDetail(erCode);
    setupResponse(call, headerBuilder.build(), null);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
/**
 * get the rpc status corresponding to this exception
 */
public RpcStatusProto getRpcStatusProto() {
  return RpcStatusProto.ERROR;
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto; //导入依赖的package包/类
private void receiveRpcResponse() {
  if (shouldCloseConnection.get()) {
    return;
  }
  touch();
  
  try {
    int totalLen = in.readInt();
    RpcResponseHeaderProto header = 
        RpcResponseHeaderProto.parseDelimitedFrom(in);
    checkResponse(header);

    int headerLen = header.getSerializedSize();
    headerLen += CodedOutputStream.computeRawVarint32Size(headerLen);

    int callId = header.getCallId();
    if (LOG.isDebugEnabled())
      LOG.debug(getName() + " got value #" + callId);

    Call call = calls.get(callId);
    RpcStatusProto status = header.getStatus();
    if (status == RpcStatusProto.SUCCESS) {
      Writable value = ReflectionUtils.newInstance(valueClass, conf);
      value.readFields(in);                 // read value
      calls.remove(callId);
      call.setRpcResponse(value);
      
      // verify that length was correct
      // only for ProtobufEngine where len can be verified easily
      if (call.getRpcResponse() instanceof ProtobufRpcEngine.RpcWrapper) {
        ProtobufRpcEngine.RpcWrapper resWrapper = 
            (ProtobufRpcEngine.RpcWrapper) call.getRpcResponse();
        if (totalLen != headerLen + resWrapper.getLength()) { 
          throw new RpcClientException(
              "RPC response length mismatch on rpc success");
        }
      }
    } else { // Rpc Request failed
      // Verify that length was correct
      if (totalLen != headerLen) {
        throw new RpcClientException(
            "RPC response length mismatch on rpc error");
      }
      
      final String exceptionClassName = header.hasExceptionClassName() ?
            header.getExceptionClassName() : 
              "ServerDidNotSetExceptionClassName";
      final String errorMsg = header.hasErrorMsg() ? 
            header.getErrorMsg() : "ServerDidNotSetErrorMsg" ;
      final RpcErrorCodeProto erCode = 
                (header.hasErrorDetail() ? header.getErrorDetail() : null);
      if (erCode == null) {
         LOG.warn("Detailed error code not set by server on rpc error");
      }
      RemoteException re = 
          ( (erCode == null) ? 
              new RemoteException(exceptionClassName, errorMsg) :
          new RemoteException(exceptionClassName, errorMsg, erCode));
      if (status == RpcStatusProto.ERROR) {
        calls.remove(callId);
        call.setException(re);
      } else if (status == RpcStatusProto.FATAL) {
        // Close the connection
        markClosed(re);
      }
    }
  } catch (IOException e) {
    markClosed(e);
  }
}