import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
/**
 * Build the user information
 *
 * @param ugi        User Group Information
 * @param authMethod Authorization method
 * @return UserInformation protobuf
 */
private RPCProtos.UserInformation buildUserInfo(UserGroupInformation ugi, AuthMethod authMethod) {
  if (ugi == null || authMethod == AuthMethod.DIGEST) {
    // Don't send user for token auth
    return null;
  }
  RPCProtos.UserInformation.Builder userInfoPB = RPCProtos.UserInformation.newBuilder();
  if (authMethod == AuthMethod.KERBEROS) {
    // Send effective user for Kerberos auth
    userInfoPB.setEffectiveUser(ugi.getUserName());
  } else if (authMethod == AuthMethod.SIMPLE) {
    //Send both effective user and real user for simple auth
    userInfoPB.setEffectiveUser(ugi.getUserName());
    if (ugi.getRealUser() != null) {
      userInfoPB.setRealUser(ugi.getRealUser().getUserName());
    }
  }
  return userInfoPB.build();
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
public int getPriority(RPCProtos.RequestHeader header, Message param, User user) {
  // Yes this is copy pasted from the base class but it keeps from having to look in the
  // annotatedQos table twice something that could get costly since this is called for
  // every single RPC request.
  int priorityByAnnotation = getAnnotatedPriority(header);
  if (priorityByAnnotation >= 0) {
    return priorityByAnnotation;
  }

  // If meta is moving then all the other of reports of state transitions will be
  // un able to edit meta. Those blocked reports should not keep the report that opens meta from
  // running. Hence all reports of meta transitioning should always be in a different thread.
  // This keeps from deadlocking the cluster.
  if (param instanceof RegionServerStatusProtos.ReportRegionStateTransitionRequest) {
    // Regions are moving. Lets see which ones.
    RegionServerStatusProtos.ReportRegionStateTransitionRequest
        tRequest = (RegionServerStatusProtos.ReportRegionStateTransitionRequest) param;
    for (RegionServerStatusProtos.RegionStateTransition rst : tRequest.getTransitionList()) {
      if (rst.getRegionInfoList() != null) {
        for (HBaseProtos.RegionInfo info : rst.getRegionInfoList()) {
          TableName tn = ProtobufUtil.toTableName(info.getTableName());
          if (tn.isSystemTable()) {
            return HConstants.SYSTEMTABLE_QOS;
          }
        }
      }
    }
    return HConstants.NORMAL_QOS;
  }

  // Handle the rest of the different reasons to change priority.
  return getBasePriority(header, param);
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
protected void checkMethod(Configuration conf, final String methodName, final int expected,
                           final AnnotationReadingPriorityFunction qosf, final Message param) {
  RPCProtos.RequestHeader.Builder builder = RPCProtos.RequestHeader.newBuilder();
  builder.setMethodName(methodName);
  assertEquals(methodName, expected, qosf.getPriority(builder.build(), param,
    User.createUserForTesting(conf, "someuser", new String[]{"somegroup"})));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
/**
 * Write the channel header
 *
 * @param channel to write to
 * @return future of write
 * @throws java.io.IOException on failure to write
 */
private ChannelFuture writeChannelHeader(Channel channel) throws IOException {
  RPCProtos.ConnectionHeader.Builder headerBuilder =
      RPCProtos.ConnectionHeader.newBuilder().setServiceName(serviceName);

  RPCProtos.UserInformation userInfoPB = buildUserInfo(ticket.getUGI(), authMethod);
  if (userInfoPB != null) {
    headerBuilder.setUserInfo(userInfoPB);
  }

  if (client.codec != null) {
    headerBuilder.setCellBlockCodecClass(client.codec.getClass().getCanonicalName());
  }
  if (client.compressor != null) {
    headerBuilder.setCellBlockCompressorClass(client.compressor.getClass().getCanonicalName());
  }

  headerBuilder.setVersionInfo(ProtobufUtil.getVersionInfo());
  RPCProtos.ConnectionHeader header = headerBuilder.build();


  int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(header);

  ByteBuf b = channel.alloc().directBuffer(totalSize);

  b.writeInt(header.getSerializedSize());
  b.writeBytes(header.toByteArray());

  return channel.writeAndFlush(b);
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
/**
 * @param e Proto exception
 * @return RemoteException made from passed <code>e</code>
 */
private RemoteException createRemoteException(final RPCProtos.ExceptionResponse e) {
  String innerExceptionClassName = e.getExceptionClassName();
  boolean doNotRetry = e.getDoNotRetry();
  return e.hasHostname() ?
      // If a hostname then add it to the RemoteWithExtrasException
      new RemoteWithExtrasException(innerExceptionClassName, e.getStackTrace(), e.getHostname(),
          e.getPort(), doNotRetry) :
      new RemoteWithExtrasException(innerExceptionClassName, e.getStackTrace(), doNotRetry);
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
/**
 * Write the channel header
 *
 * @param channel to write to
 * @return future of write
 * @throws java.io.IOException on failure to write
 */
private ChannelFuture writeChannelHeader(Channel channel) throws IOException {
  RPCProtos.ConnectionHeader.Builder headerBuilder =
      RPCProtos.ConnectionHeader.newBuilder().setServiceName(serviceName);

  RPCProtos.ConnectionHeader.Builder builder = RPCProtos.ConnectionHeader.newBuilder();
  builder.setServiceName(serviceName);
  RPCProtos.UserInformation userInfoPB = buildUserInfo(ticket.getUGI(), authMethod);
  if (userInfoPB != null) {
    headerBuilder.setUserInfo(userInfoPB);
  }

  if (client.codec != null) {
    headerBuilder.setCellBlockCodecClass(client.codec.getClass().getCanonicalName());
  }
  if (client.compressor != null) {
    headerBuilder.setCellBlockCompressorClass(client.compressor.getClass().getCanonicalName());
  }

  RPCProtos.ConnectionHeader header = headerBuilder.build();


  int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(header);

  ByteBuf b = channel.alloc().directBuffer(totalSize);

  b.writeInt(header.getSerializedSize());
  b.writeBytes(header.toByteArray());

  return channel.writeAndFlush(b);
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
protected void sendParam(Call call) {
  if (shouldCloseConnection.get()) {
    return;
  }
  try {
    if (LOG.isDebugEnabled())
      LOG.debug(getName() + " sending #" + call.id);

    RpcRequestHeader.Builder headerBuilder = RPCProtos.RpcRequestHeader.newBuilder();
    headerBuilder.setCallId(call.id);

    if (Trace.isTracing()) {
      Span s = Trace.currentTrace();
      headerBuilder.setTinfo(RPCTInfo.newBuilder()
          .setParentId(s.getSpanId())
          .setTraceId(s.getTraceId()));
    }

    //noinspection SynchronizeOnNonFinalField
    synchronized (this.out) { // FindBugs IS2_INCONSISTENT_SYNC
      RpcRequestHeader header = headerBuilder.build();
      int serializedHeaderSize = header.getSerializedSize();
      int requestSerializedSize = call.param.getSerializedSize();
      this.out.writeInt(serializedHeaderSize +
          CodedOutputStream.computeRawVarint32Size(serializedHeaderSize) +
          requestSerializedSize +
          CodedOutputStream.computeRawVarint32Size(requestSerializedSize));
      header.writeDelimitedTo(this.out);
      call.param.writeDelimitedTo(this.out);
      this.out.flush();
    }
  } catch(IOException e) {
    markClosed(e);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
/**
 * Write request to channel
 *
 * @param call    to write
 */
private void writeRequest(final AsyncCall call) {
  try {
    final RPCProtos.RequestHeader.Builder requestHeaderBuilder = RPCProtos.RequestHeader
        .newBuilder();
    requestHeaderBuilder.setCallId(call.id)
            .setMethodName(call.method.getName()).setRequestParam(call.param != null);

    if (Trace.isTracing()) {
      Span s = Trace.currentSpan();
      requestHeaderBuilder.setTraceInfo(TracingProtos.RPCTInfo.newBuilder().
          setParentId(s.getSpanId()).setTraceId(s.getTraceId()));
    }

    ByteBuffer cellBlock = client.buildCellBlock(call.controller.cellScanner());
    if (cellBlock != null) {
      final RPCProtos.CellBlockMeta.Builder cellBlockBuilder = RPCProtos.CellBlockMeta
          .newBuilder();
      cellBlockBuilder.setLength(cellBlock.limit());
      requestHeaderBuilder.setCellBlockMeta(cellBlockBuilder.build());
    }
    // Only pass priority if there one.  Let zero be same as no priority.
    if (call.controller.getPriority() != 0) {
      requestHeaderBuilder.setPriority(call.controller.getPriority());
    }

    RPCProtos.RequestHeader rh = requestHeaderBuilder.build();

    int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(rh, call.param);
    if (cellBlock != null) {
      totalSize += cellBlock.remaining();
    }

    ByteBuf b = channel.alloc().directBuffer(4 + totalSize);
    try(ByteBufOutputStream out = new ByteBufOutputStream(b)) {
      call.callStats.setRequestSizeBytes(IPCUtil.write(out, rh, call.param, cellBlock));
    }

    channel.writeAndFlush(b).addListener(new CallWriteListener(this, call.id));
  } catch (IOException e) {
    close(e);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  ByteBuf inBuffer = (ByteBuf) msg;
  ByteBufInputStream in = new ByteBufInputStream(inBuffer);
  int totalSize = inBuffer.readableBytes();
  try {
    // Read the header
    RPCProtos.ResponseHeader responseHeader = RPCProtos.ResponseHeader.parseDelimitedFrom(in);
    int id = responseHeader.getCallId();
    AsyncCall call = channel.removePendingCall(id);
    if (call == null) {
      // So we got a response for which we have no corresponding 'call' here on the client-side.
      // We probably timed out waiting, cleaned up all references, and now the server decides
      // to return a response.  There is nothing we can do w/ the response at this stage. Clean
      // out the wire of the response so its out of the way and we can get other responses on
      // this connection.
      int readSoFar = IPCUtil.getTotalSizeWhenWrittenDelimited(responseHeader);
      int whatIsLeftToRead = totalSize - readSoFar;

      // This is done through a Netty ByteBuf which has different behavior than InputStream.
      // It does not return number of bytes read but will update pointer internally and throws an
      // exception when too many bytes are to be skipped.
      inBuffer.skipBytes(whatIsLeftToRead);
      return;
    }

    if (responseHeader.hasException()) {
      RPCProtos.ExceptionResponse exceptionResponse = responseHeader.getException();
      RemoteException re = createRemoteException(exceptionResponse);
      if (exceptionResponse.getExceptionClassName().
          equals(FatalConnectionException.class.getName())) {
        channel.close(re);
      } else {
        call.setFailed(re);
      }
    } else {
      Message value = null;
      // Call may be null because it may have timedout and been cleaned up on this side already
      if (call.responseDefaultType != null) {
        Message.Builder builder = call.responseDefaultType.newBuilderForType();
        ProtobufUtil.mergeDelimitedFrom(builder, in);
        value = builder.build();
      }
      CellScanner cellBlockScanner = null;
      if (responseHeader.hasCellBlockMeta()) {
        int size = responseHeader.getCellBlockMeta().getLength();
        byte[] cellBlock = new byte[size];
        inBuffer.readBytes(cellBlock, 0, cellBlock.length);
        cellBlockScanner = channel.client.createCellScanner(cellBlock);
      }
      call.setSuccess(value, cellBlockScanner);
      call.callStats.setResponseSizeBytes(totalSize);
    }
  } catch (IOException e) {
    // Treat this as a fatal condition and close this connection
    channel.close(e);
  } finally {
    inBuffer.release();
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
@Override
public int getPriority(RPCProtos.RequestHeader header, Message param) {
  return 0;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
/**
 * Write request to channel
 *
 * @param call    to write
 */
private void writeRequest(final AsyncCall call) {
  try {
    if (shouldCloseConnection) {
      return;
    }

    final RPCProtos.RequestHeader.Builder requestHeaderBuilder = RPCProtos.RequestHeader
        .newBuilder();
    requestHeaderBuilder.setCallId(call.id)
        .setMethodName(call.method.getName()).setRequestParam(call.param != null);

    if (Trace.isTracing()) {
      Span s = Trace.currentSpan();
      requestHeaderBuilder.setTraceInfo(TracingProtos.RPCTInfo.newBuilder().
          setParentId(s.getSpanId()).setTraceId(s.getTraceId()));
    }

    ByteBuffer cellBlock = client.buildCellBlock(call.controller.cellScanner());
    if (cellBlock != null) {
      final RPCProtos.CellBlockMeta.Builder cellBlockBuilder = RPCProtos.CellBlockMeta
          .newBuilder();
      cellBlockBuilder.setLength(cellBlock.limit());
      requestHeaderBuilder.setCellBlockMeta(cellBlockBuilder.build());
    }
    // Only pass priority if there one.  Let zero be same as no priority.
    if (call.controller.getPriority() != 0) {
      requestHeaderBuilder.setPriority(call.controller.getPriority());
    }

    RPCProtos.RequestHeader rh = requestHeaderBuilder.build();

    int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(rh, call.param);
    if (cellBlock != null) {
      totalSize += cellBlock.remaining();
    }

    ByteBuf b = channel.alloc().directBuffer(totalSize);
    try(ByteBufOutputStream out = new ByteBufOutputStream(b)) {
      IPCUtil.write(out, rh, call.param, cellBlock);
    }

    channel.writeAndFlush(b).addListener(new CallWriteListener(this,call));
  } catch (IOException e) {
    if (!shouldCloseConnection) {
      close(e);
    }
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; //导入依赖的package包/类
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  ByteBuf inBuffer = (ByteBuf) msg;
  ByteBufInputStream in = new ByteBufInputStream(inBuffer);

  if (channel.shouldCloseConnection) {
    return;
  }
  int totalSize = inBuffer.readableBytes();
  try {
    // Read the header
    RPCProtos.ResponseHeader responseHeader = RPCProtos.ResponseHeader.parseDelimitedFrom(in);
    int id = responseHeader.getCallId();
    AsyncCall call = channel.calls.get(id);
    if (call == null) {
      // So we got a response for which we have no corresponding 'call' here on the client-side.
      // We probably timed out waiting, cleaned up all references, and now the server decides
      // to return a response.  There is nothing we can do w/ the response at this stage. Clean
      // out the wire of the response so its out of the way and we can get other responses on
      // this connection.
      int readSoFar = IPCUtil.getTotalSizeWhenWrittenDelimited(responseHeader);
      int whatIsLeftToRead = totalSize - readSoFar;

      // This is done through a Netty ByteBuf which has different behavior than InputStream.
      // It does not return number of bytes read but will update pointer internally and throws an
      // exception when too many bytes are to be skipped.
      inBuffer.skipBytes(whatIsLeftToRead);
      return;
    }

    if (responseHeader.hasException()) {
      RPCProtos.ExceptionResponse exceptionResponse = responseHeader.getException();
      RemoteException re = createRemoteException(exceptionResponse);
      if (exceptionResponse.getExceptionClassName().
          equals(FatalConnectionException.class.getName())) {
        channel.close(re);
      } else {
        channel.failCall(call, re);
      }
    } else {
      Message value = null;
      // Call may be null because it may have timedout and been cleaned up on this side already
      if (call.responseDefaultType != null) {
        Message.Builder builder = call.responseDefaultType.newBuilderForType();
        builder.mergeDelimitedFrom(in);
        value = builder.build();
      }
      CellScanner cellBlockScanner = null;
      if (responseHeader.hasCellBlockMeta()) {
        int size = responseHeader.getCellBlockMeta().getLength();
        byte[] cellBlock = new byte[size];
        inBuffer.readBytes(cellBlock, 0, cellBlock.length);
        cellBlockScanner = channel.client.createCellScanner(cellBlock);
      }
      call.setSuccess(value, cellBlockScanner);
    }
    channel.calls.remove(id);
  } catch (IOException e) {
    // Treat this as a fatal condition and close this connection
    channel.close(e);
  } finally {
    inBuffer.release();
    channel.cleanupCalls(false);
  }
}