import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * A helper to get a row of the closet one before using client protocol.
 *
 * @param client
 * @param regionName
 * @param row
 * @param family
 * @return the row or the closestRowBefore if it doesn't exist
 * @throws IOException
 * @deprecated since 0.99 - use reversed scanner instead.
 */
@Deprecated
public static Result getRowOrBefore(final ClientService.BlockingInterface client,
    final byte[] regionName, final byte[] row,
    final byte[] family) throws IOException {
  GetRequest request =
    RequestConverter.buildGetRowOrBeforeRequest(
      regionName, row, family);
  try {
    GetResponse response = client.get(null, request);
    if (!response.hasResult()) return null;
    return toResult(response.getResult());
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * Make a region server endpoint call
 * @param client
 * @param call
 * @return CoprocessorServiceResponse
 * @throws IOException
 */
public static CoprocessorServiceResponse execRegionServerService(
    final RpcController controller, final ClientService.BlockingInterface client,
    final CoprocessorServiceCall call)
    throws IOException {
  CoprocessorServiceRequest request =
      CoprocessorServiceRequest
          .newBuilder()
          .setCall(call)
          .setRegion(
            RequestConverter.buildRegionSpecifier(REGION_NAME, HConstants.EMPTY_BYTE_ARRAY))
          .build();
  try {
    CoprocessorServiceResponse response = client.execRegionServerService(controller, request);
    return response;
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
throws IOException {
  if (isDeadServer(sn)) {
    throw new RegionServerStoppedException(sn + " is dead.");
  }
  String key = getStubKey(ClientService.BlockingInterface.class.getName(), sn.getHostname(),
      sn.getPort(), this.hostnamesCanChange);
  this.connectionLock.putIfAbsent(key, key);
  ClientService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (ClientService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel =
          this.rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
      stub = ClientService.newBlockingStub(channel);
      // In old days, after getting stub/proxy, we'd make a call.  We are not doing that here.
      // Just fail on first actual call rather than in here on setup.
      this.stubs.put(key, stub);
    }
  }
  return stub;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
ScanOpenNextThenExceptionThenRecoverConnection(Configuration conf,
    boolean managed, ExecutorService pool) throws IOException {
  super(conf, managed);
  // Mock up my stub so open scanner returns a scanner id and then on next, we throw
  // exceptions for three times and then after that, we return no more to scan.
  this.stub = Mockito.mock(ClientService.BlockingInterface.class);
  long sid = 12345L;
  try {
    Mockito.when(stub.scan((RpcController)Mockito.any(),
        (ClientProtos.ScanRequest)Mockito.any())).
      thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).build()).
      thenThrow(new ServiceException(new RegionServerStoppedException("From Mockito"))).
      thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).
          setMoreResults(false).build());
  } catch (ServiceException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
RegionServerStoppedOnScannerOpenConnection(Configuration conf, boolean managed,
    ExecutorService pool, User user) throws IOException {
  super(conf, managed);
  // Mock up my stub so open scanner returns a scanner id and then on next, we throw
  // exceptions for three times and then after that, we return no more to scan.
  this.stub = Mockito.mock(ClientService.BlockingInterface.class);
  long sid = 12345L;
  try {
    Mockito.when(stub.scan((RpcController)Mockito.any(),
        (ClientProtos.ScanRequest)Mockito.any())).
      thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).build()).
      thenThrow(new ServiceException(new RegionServerStoppedException("From Mockito"))).
      thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).
          setMoreResults(false).build());
  } catch (ServiceException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * Make a region server endpoint call
 * @param client
 * @param call
 * @return CoprocessorServiceResponse
 * @throws IOException
 */
public static CoprocessorServiceResponse execRegionServerService(
    final ClientService.BlockingInterface client, final CoprocessorServiceCall call)
    throws IOException {
  CoprocessorServiceRequest request =
      CoprocessorServiceRequest
          .newBuilder()
          .setCall(call)
          .setRegion(
            RequestConverter.buildRegionSpecifier(REGION_NAME, HConstants.EMPTY_BYTE_ARRAY))
          .build();
  try {
    CoprocessorServiceResponse response = client.execRegionServerService(null, request);
    return response;
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
        throws IOException {
    if (isDeadServer(sn)) {
        throw new RegionServerStoppedException(sn + " is dead.");
    }
    String key = getStubKey(ClientService.BlockingInterface.class.getName(), sn.getHostAndPort());
    this.connectionLock.putIfAbsent(key, key);
    ClientService.BlockingInterface stub = null;
    synchronized (this.connectionLock.get(key)) {
        stub = (ClientService.BlockingInterface) this.stubs.get(key);
        if (stub == null) {
            BlockingRpcChannel channel =
                    this.rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
            stub = ClientService.newBlockingStub(channel);
            // In old days, after getting stub/proxy, we'd make a call.  We are not doing that here.
            // Just fail on first actual call rather than in here on setup.
            this.stubs.put(key, stub);
        }
    }
    return stub;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
throws IOException {
  if (isDeadServer(sn)) {
    throw new RegionServerStoppedException(sn + " is dead.");
  }
  String key = getStubKey(ClientService.BlockingInterface.class.getName(), sn.getHostAndPort());
  this.connectionLock.putIfAbsent(key, key);
  ClientService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (ClientService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel = this.rpcClient.createBlockingRpcChannel(sn,
        user, this.rpcTimeout);
      stub = ClientService.newBlockingStub(channel);
      // In old days, after getting stub/proxy, we'd make a call.  We are not doing that here.
      // Just fail on first actual call rather than in here on setup.
      this.stubs.put(key, stub);
    }
  }
  return stub;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
@Override
public ClientService.BlockingInterface getClient(final ServerName sn)
throws IOException {
  if (isDeadServer(sn)) {
    throw new RegionServerStoppedException(sn + " is dead.");
  }
  String key = getStubKey(ClientService.BlockingInterface.class.getName(), sn.getHostAndPort());
  this.connectionLock.putIfAbsent(key, key);
  ClientService.BlockingInterface stub = null;
  synchronized (this.connectionLock.get(key)) {
    stub = (ClientService.BlockingInterface)this.stubs.get(key);
    if (stub == null) {
      BlockingRpcChannel channel =
          this.rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
      stub = ClientService.newBlockingStub(channel);
      // In old days, after getting stub/proxy, we'd make a call.  We are not doing that here.
      // Just fail on first actual call rather than in here on setup.
      this.stubs.put(key, stub);
    }
  }
  return stub;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * @return list of blocking services and their security info classes that this server supports
 */
protected List<BlockingServiceAndInterface> getServices() {
  List<BlockingServiceAndInterface> bssi = new ArrayList<BlockingServiceAndInterface>(2);
  bssi.add(new BlockingServiceAndInterface(
    ClientService.newReflectiveBlockingService(this),
    ClientService.BlockingInterface.class));
  bssi.add(new BlockingServiceAndInterface(
    AdminService.newReflectiveBlockingService(this),
    AdminService.BlockingInterface.class));
  return bssi;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
@Test
@SuppressWarnings("deprecation")
public void testShortCircuitConnection() throws IOException, InterruptedException {
  String tnAsString = "testShortCircuitConnection";
  TableName tn = TableName.valueOf(tnAsString);
  HTableDescriptor htd = UTIL.createTableDescriptor(tnAsString);
  HColumnDescriptor hcd = new HColumnDescriptor(Bytes.toBytes("cf"));
  htd.addFamily(hcd);
  UTIL.createTable(htd, null);
  HRegionServer regionServer = UTIL.getRSForFirstRegionInTable(tn);
  ClusterConnection connection = regionServer.getConnection();
  HTableInterface tableIf = connection.getTable(tn);
  assertTrue(tableIf instanceof HTable);
  HTable table = (HTable) tableIf;
  assertTrue(table.getConnection() == connection);
  AdminService.BlockingInterface admin = connection.getAdmin(regionServer.getServerName());
  ClientService.BlockingInterface client = connection.getClient(regionServer.getServerName());
  assertTrue(admin instanceof RSRpcServices);
  assertTrue(client instanceof RSRpcServices);
  ServerName anotherSn = ServerName.valueOf(regionServer.getServerName().getHostAndPort(),
    EnvironmentEdgeManager.currentTime());
  admin = connection.getAdmin(anotherSn);
  client = connection.getClient(anotherSn);
  assertFalse(admin instanceof RSRpcServices);
  assertFalse(client instanceof RSRpcServices);
  assertTrue(connection.getAdmin().getConnection() == connection);
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * A helper to bulk load a list of HFiles using client protocol.
 *
 * @param client
 * @param familyPaths
 * @param regionName
 * @param assignSeqNum
 * @return true if all are loaded
 * @throws IOException
 */
public static boolean bulkLoadHFile(final ClientService.BlockingInterface client,
    final List<Pair<byte[], String>> familyPaths,
    final byte[] regionName, boolean assignSeqNum) throws IOException {
  BulkLoadHFileRequest request =
    RequestConverter.buildBulkLoadHFileRequest(familyPaths, regionName, assignSeqNum);
  try {
    BulkLoadHFileResponse response =
      client.bulkLoadHFile(null, request);
    return response.getLoaded();
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
public static CoprocessorServiceResponse execService(final RpcController controller,
    final ClientService.BlockingInterface client, final CoprocessorServiceCall call,
    final byte[] regionName) throws IOException {
  CoprocessorServiceRequest request = CoprocessorServiceRequest.newBuilder()
      .setCall(call).setRegion(
          RequestConverter.buildRegionSpecifier(REGION_NAME, regionName)).build();
  try {
    CoprocessorServiceResponse response =
        client.execService(controller, request);
    return response;
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
RpcTimeoutConnection(Configuration conf, boolean managed, ExecutorService pool, User user)
throws IOException {
  super(conf, managed);
  // Mock up my stub so an exists call -- which turns into a get -- throws an exception
  this.stub = Mockito.mock(ClientService.BlockingInterface.class);
  try {
    Mockito.when(stub.get((RpcController)Mockito.any(),
        (ClientProtos.GetRequest)Mockito.any())).
      thenThrow(new ServiceException(new RegionServerStoppedException("From Mockito")));
  } catch (ServiceException e) {
    throw new IOException(e);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
ManyServersManyRegionsConnection(Configuration conf, boolean managed,
    ExecutorService pool, User user)
throws IOException {
  super(conf, managed, pool, user);
  int serverCount = conf.getInt("hbase.test.servers", 10);
  this.serversByClient =
    new HashMap<ServerName, ClientService.BlockingInterface>(serverCount);
  this.meta = makeMeta(Bytes.toBytes(
    conf.get("hbase.test.tablename", Bytes.toString(BIG_USER_TABLE))),
    conf.getInt("hbase.test.regions", 100),
    conf.getLong("hbase.test.namespace.span", 1000),
    serverCount);
  this.conf = conf;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
@Override
public ClientService.BlockingInterface getClient(ServerName sn) throws IOException {
  // if (!sn.toString().startsWith("meta")) LOG.info(sn);
  ClientService.BlockingInterface stub = null;
  synchronized (this.serversByClient) {
    stub = this.serversByClient.get(sn);
    if (stub == null) {
      stub = new FakeServer(this.conf, meta, sequenceids);
      this.serversByClient.put(sn, stub);
    }
  }
  return stub;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * @return list of blocking services and their security info classes that this server supports
 */
protected List<BlockingServiceAndInterface> getServices() {
    List<BlockingServiceAndInterface> bssi = new ArrayList<BlockingServiceAndInterface>(2);
    bssi.add(new BlockingServiceAndInterface(
            ClientService.newReflectiveBlockingService(this),
            ClientService.BlockingInterface.class));
    bssi.add(new BlockingServiceAndInterface(
            AdminService.newReflectiveBlockingService(this),
            AdminService.BlockingInterface.class));
    return bssi;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
public static CoprocessorServiceResponse execService(final ClientService.BlockingInterface client,
    final CoprocessorServiceCall call, final byte[] regionName) throws IOException {
  CoprocessorServiceRequest request = CoprocessorServiceRequest.newBuilder()
      .setCall(call).setRegion(
          RequestConverter.buildRegionSpecifier(REGION_NAME, regionName)).build();
  try {
    CoprocessorServiceResponse response =
        client.execService(null, request);
    return response;
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
public org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService.BlockingInterface
    getClient(ServerName serverName) throws IOException {
  // client is trying to reach off-server, so we can't do anything special
  if (!this.serverName.equals(serverName)) {
    return delegate.getClient(serverName);
  }
  // the client is attempting to write to the same regionserver, we can short-circuit to our
  // local regionserver
  final BlockingService blocking = ClientService.newReflectiveBlockingService(this.server);
  final RpcServerInterface rpc = this.server.getRpcServer();

  final MonitoredRPCHandler status =
      TaskMonitor.get().createRPCStatus(Thread.currentThread().getName());
  status.pause("Setting up server-local call");

  final long timestamp = EnvironmentEdgeManager.currentTimeMillis();
  BlockingRpcChannel channel = new BlockingRpcChannel() {

    @Override
    public Message callBlockingMethod(MethodDescriptor method, RpcController controller,
        Message request, Message responsePrototype) throws ServiceException {
      try {
        // we never need a cell-scanner - everything is already fully formed
        return rpc.call(blocking, method, request, null, timestamp, status).getFirst();
      } catch (IOException e) {
        throw new ServiceException(e);
      }
    }
  };
  return ClientService.newBlockingStub(channel);
}
 

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; //导入依赖的package包/类
/**
 * A helper to get a row of the closet one before using client protocol.
 * @param client
 * @param regionName
 * @param row
 * @param family
 * @param payloadCarryingRpcController
 * @return the row or the closestRowBefore if it doesn't exist
 * @throws IOException
 */
public static Result getRowOrBefore(final ClientService.BlockingInterface client,
    final byte[] regionName, final byte[] row, final byte[] family,
    PayloadCarryingRpcController payloadCarryingRpcController) throws IOException {
  GetRequest request =
    RequestConverter.buildGetRowOrBeforeRequest(
      regionName, row, family);
  try {
    GetResponse response = client.get(payloadCarryingRpcController, request);
    if (!response.hasResult()) return null;
    return toResult(response.getResult());
  } catch (ServiceException se) {
    throw getRemoteException(se);
  }
}