import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
public TokenServer(Configuration conf) throws IOException {
  this.conf = conf;
  this.startcode = EnvironmentEdgeManager.currentTime();
  // Server to handle client requests.
  String hostname =
    Strings.domainNamePointerToHostName(DNS.getDefaultHost("default", "default"));
  int port = 0;
  // Creation of an ISA will force a resolve.
  InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
  if (initialIsa.getAddress() == null) {
    throw new IllegalArgumentException("Failed resolve of " + initialIsa);
  }
  final List<BlockingServiceAndInterface> sai =
    new ArrayList<BlockingServiceAndInterface>(1);
  BlockingService service =
    AuthenticationProtos.AuthenticationService.newReflectiveBlockingService(this);
  sai.add(new BlockingServiceAndInterface(service,
    AuthenticationProtos.AuthenticationService.BlockingInterface.class));
  this.rpcServer =
    new RpcServer(this, "tokenServer", sai, initialIsa, conf, new FifoRpcScheduler(conf, 1));
  this.isa = this.rpcServer.getListenerAddress();
  this.sleeper = new Sleeper(1000, this);
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
public TokenServer(Configuration conf) throws IOException {
  this.conf = conf;
  this.startcode = EnvironmentEdgeManager.currentTimeMillis();
  // Server to handle client requests.
  String hostname =
    Strings.domainNamePointerToHostName(DNS.getDefaultHost("default", "default"));
  int port = 0;
  // Creation of an ISA will force a resolve.
  InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
  if (initialIsa.getAddress() == null) {
    throw new IllegalArgumentException("Failed resolve of " + initialIsa);
  }
  final List<BlockingServiceAndInterface> sai =
    new ArrayList<BlockingServiceAndInterface>(1);
  BlockingService service =
    AuthenticationProtos.AuthenticationService.newReflectiveBlockingService(this);
  sai.add(new BlockingServiceAndInterface(service,
    AuthenticationProtos.AuthenticationService.BlockingInterface.class));
  this.rpcServer =
    new RpcServer(this, "tokenServer", sai, initialIsa, conf, new FifoRpcScheduler(conf, 1));
  this.isa = this.rpcServer.getListenerAddress();
  this.sleeper = new Sleeper(1000, this);
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
/**
 * By default, put up an Admin and a Client Service.
 * Set booleans <code>hbase.regionserver.admin.executorService</code> and
 * <code>hbase.regionserver.client.executorService</code> if you want to enable/disable services.
 * Default is that both are enabled.
 * @return immutable list of blocking services and the security info classes that this server
 * supports
 */
protected List<BlockingServiceAndInterface> getServices() {
  boolean admin =
    getConfiguration().getBoolean(REGIONSERVER_ADMIN_SERVICE_CONFIG, true);
  boolean client =
    getConfiguration().getBoolean(REGIONSERVER_CLIENT_SERVICE_CONFIG, true);
  List<BlockingServiceAndInterface> bssi = new ArrayList<>();
  if (client) {
    bssi.add(new BlockingServiceAndInterface(
    ClientService.newReflectiveBlockingService(this),
    ClientService.BlockingInterface.class));
  }
  if (admin) {
    bssi.add(new BlockingServiceAndInterface(
    AdminService.newReflectiveBlockingService(this),
    AdminService.BlockingInterface.class));
  }
  return new org.apache.hbase.thirdparty.com.google.common.collect.
      ImmutableList.Builder<BlockingServiceAndInterface>().addAll(bssi).build();
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
public static RpcServer createRpcServer(final Server server, final String name,
    final List<BlockingServiceAndInterface> services,
    final InetSocketAddress bindAddress, Configuration conf,
    RpcScheduler scheduler, boolean reservoirEnabled) throws IOException {
  String rpcServerClass = conf.get(CUSTOM_RPC_SERVER_IMPL_CONF_KEY,
      NettyRpcServer.class.getName());
  StringBuilder servicesList = new StringBuilder();
  for (BlockingServiceAndInterface s: services) {
    ServiceDescriptor sd = s.getBlockingService().getDescriptorForType();
    if (sd == null) continue; // Can be null for certain tests like TestTokenAuthentication
    if (servicesList.length() > 0) servicesList.append(", ");
    servicesList.append(sd.getFullName());
  }
  LOG.info("Creating " + rpcServerClass + " hosting " + servicesList);
  return ReflectionUtils.instantiateWithCustomCtor(rpcServerClass,
      new Class[] { Server.class, String.class, List.class,
        InetSocketAddress.class, Configuration.class, RpcScheduler.class, boolean.class },
      new Object[] { server, name, services, bindAddress, conf, scheduler, reservoirEnabled });
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
@Ignore
@Test
public void testRpcScheduler() throws IOException, InterruptedException {
  PriorityFunction qosFunction = mock(PriorityFunction.class);
  Abortable abortable = new AbortServer();
  CONF.set(RpcServerFactory.CUSTOM_RPC_SERVER_IMPL_CONF_KEY, rpcServerImpl);
  RpcScheduler scheduler = new SimpleRpcScheduler(CONF, 2, 0, 0, qosFunction, abortable, 0);
  RpcServer rpcServer = RpcServerFactory.createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new BlockingServiceAndInterface((BlockingService) SERVICE, null)),
      new InetSocketAddress("localhost", 0), CONF, scheduler);
  try (BlockingRpcClient client = new BlockingRpcClient(CONF)) {
    rpcServer.start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    stub.echo(null, EchoRequestProto.newBuilder().setMessage("hello").build());
  } catch (Throwable e) {
    assert (abortable.isAborted() == true);
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
/**
 * Ensure we do not HAVE TO HAVE a codec.
 */
@Test
public void testNoCodec() throws IOException, ServiceException {
  Configuration conf = HBaseConfiguration.create();
  RpcServer rpcServer = createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF,
      new FifoRpcScheduler(CONF, 1));
  try (AbstractRpcClient<?> client = createRpcClientNoCodec(conf)) {
    rpcServer.start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    HBaseRpcController pcrc = new HBaseRpcControllerImpl();
    String message = "hello";
    assertEquals(message,
      stub.echo(pcrc, EchoRequestProto.newBuilder().setMessage(message).build()).getMessage());
    assertNull(pcrc.cellScanner());
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
@Test
public void testRTEDuringConnectionSetup() throws Exception {
  Configuration conf = HBaseConfiguration.create();
  RpcServer rpcServer = createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF,
      new FifoRpcScheduler(CONF, 1));
  try (AbstractRpcClient<?> client = createRpcClientRTEDuringConnectionSetup(conf)) {
    rpcServer.start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    stub.ping(null, EmptyRequestProto.getDefaultInstance());
    fail("Expected an exception to have been thrown!");
  } catch (Exception e) {
    LOG.info("Caught expected exception: " + e.toString());
    assertTrue(e.toString(), StringUtils.stringifyException(e).contains("Injected fault"));
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
/**
 * Tests that the rpc scheduler is called when requests arrive.
 */
@Test
public void testRpcScheduler() throws IOException, ServiceException, InterruptedException {
  RpcScheduler scheduler = spy(new FifoRpcScheduler(CONF, 1));
  RpcServer rpcServer = createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF, scheduler);
  verify(scheduler).init((RpcScheduler.Context) anyObject());
  try (AbstractRpcClient<?> client = createRpcClient(CONF)) {
    rpcServer.start();
    verify(scheduler).start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
    for (int i = 0; i < 10; i++) {
      stub.echo(null, param);
    }
    verify(scheduler, times(10)).dispatch((CallRunner) anyObject());
  } finally {
    rpcServer.stop();
    verify(scheduler).stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
/**
 * Tests that the RpcServer creates & dispatches CallRunner object to scheduler with non-null
 * remoteAddress set to its Call Object
 * @throws ServiceException
 */
@Test
public void testRpcServerForNotNullRemoteAddressInCallObject()
    throws IOException, ServiceException {
  RpcServer rpcServer = createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF,
      new FifoRpcScheduler(CONF, 1));
  InetSocketAddress localAddr = new InetSocketAddress("localhost", 0);
  try (AbstractRpcClient<?> client = createRpcClient(CONF)) {
    rpcServer.start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    assertEquals(localAddr.getAddress().getHostAddress(),
      stub.addr(null, EmptyRequestProto.getDefaultInstance()).getAddr());
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
@Test
public void testRemoteError() throws IOException, ServiceException {
  RpcServer rpcServer = createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF,
      new FifoRpcScheduler(CONF, 1));
  try (AbstractRpcClient<?> client = createRpcClient(CONF)) {
    rpcServer.start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    stub.error(null, EmptyRequestProto.getDefaultInstance());
  } catch (ServiceException e) {
    LOG.info("Caught expected exception: " + e);
    IOException ioe = ProtobufUtil.handleRemoteException(e);
    assertTrue(ioe instanceof DoNotRetryIOException);
    assertTrue(ioe.getMessage().contains("server error!"));
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
/** Tests that the connection closing is handled by the client with outstanding RPC calls */
@Test
public void testConnectionCloseWithOutstandingRPCs() throws InterruptedException, IOException {
  Configuration conf = new Configuration(CONF);
  RpcServer rpcServer = createTestFailingRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF,
      new FifoRpcScheduler(CONF, 1));

  try (AbstractRpcClient<?> client = createRpcClient(conf)) {
    rpcServer.start();
    BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
    EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
    stub.echo(null, param);
    fail("RPC should have failed because connection closed");
  } catch (ServiceException e) {
    LOG.info("Caught expected exception: " + e.toString());
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
@Test
public void testAsyncRemoteError() throws IOException {
  AbstractRpcClient<?> client = createRpcClient(CONF);
  RpcServer rpcServer = createRpcServer(null, "testRpcServer",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(
          SERVICE, null)), new InetSocketAddress("localhost", 0), CONF,
      new FifoRpcScheduler(CONF, 1));
  try {
    rpcServer.start();
    Interface stub = newStub(client, rpcServer.getListenerAddress());
    BlockingRpcCallback<EmptyResponseProto> callback = new BlockingRpcCallback<>();
    HBaseRpcController pcrc = new HBaseRpcControllerImpl();
    stub.error(pcrc, EmptyRequestProto.getDefaultInstance(), callback);
    assertNull(callback.get());
    assertTrue(pcrc.failed());
    LOG.info("Caught expected exception: " + pcrc.getFailed());
    IOException ioe = ProtobufUtil.handleRemoteException(pcrc.getFailed());
    assertTrue(ioe instanceof DoNotRetryIOException);
    assertTrue(ioe.getMessage().contains("server error!"));
  } finally {
    client.close();
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
RpcServer startServer() throws IOException {
  lock.writeLock().lock();
  try {
    if (rpcServers.size() >= maxServers) {
      return null;
    }

    RpcServer rpcServer = RpcServerFactory.createRpcServer(null,
        "testRpcServer", Lists
            .newArrayList(new BlockingServiceAndInterface(SERVICE, null)),
        new InetSocketAddress("localhost", 0), conf, new FifoRpcScheduler(
            conf, 1));
    rpcServer.start();
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    rpcServers.put(address, rpcServer);
    serverList.add(rpcServer);
    LOG.info("Started server: " + address);
    return rpcServer;
  } finally {
    lock.writeLock().unlock();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
public TokenServer(Configuration conf) throws IOException {
  this.conf = conf;
  this.startcode = EnvironmentEdgeManager.currentTimeMillis();
  // Server to handle client requests.
  String hostname =
    Strings.domainNamePointerToHostName(DNS.getDefaultHost("default", "default"));
  int port = 0;
  // Creation of an ISA will force a resolve.
  InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
  if (initialIsa.getAddress() == null) {
    throw new IllegalArgumentException("Failed resolve of " + initialIsa);
  }
  final List<BlockingServiceAndInterface> sai =
    new ArrayList<BlockingServiceAndInterface>(1);
  BlockingService service =
    AuthenticationProtos.AuthenticationService.newReflectiveBlockingService(this);
  sai.add(new BlockingServiceAndInterface(service,
    AuthenticationProtos.AuthenticationService.BlockingInterface.class));
  this.rpcServer =
    new RpcServer(this, "tokenServer", sai, initialIsa, 3, 1, conf, HConstants.QOS_THRESHOLD);
  this.isa = this.rpcServer.getListenerAddress();
  this.sleeper = new Sleeper(1000, this);
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的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>(4);
  bssi.add(new BlockingServiceAndInterface(
    MasterService.newReflectiveBlockingService(this),
    MasterService.BlockingInterface.class));
  bssi.add(new BlockingServiceAndInterface(
    RegionServerStatusService.newReflectiveBlockingService(this),
    RegionServerStatusService.BlockingInterface.class));
  bssi.addAll(super.getServices());
  return bssi;
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的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.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
public TokenServer(Configuration conf) throws IOException {
  this.conf = conf;
  this.startcode = EnvironmentEdgeManager.currentTime();
  // Server to handle client requests.
  String hostname =
    Strings.domainNamePointerToHostName(DNS.getDefaultHost("default", "default"));
  int port = 0;
  // Creation of an ISA will force a resolve.
  InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
  if (initialIsa.getAddress() == null) {
    throw new IllegalArgumentException("Failed resolve of " + initialIsa);
  }
  final List<BlockingServiceAndInterface> sai =
    new ArrayList<BlockingServiceAndInterface>(1);
  BlockingService service =
    AuthenticationProtos.AuthenticationService.newReflectiveBlockingService(this);
  sai.add(new BlockingServiceAndInterface(service,
    AuthenticationProtos.AuthenticationService.BlockingInterface.class));
  this.rpcServer =
    new RpcServer(this, "tokenServer", sai, initialIsa, conf, new FifoRpcScheduler(conf, 1));
  InetSocketAddress address = rpcServer.getListenerAddress();
  if (address == null) {
    throw new IOException("Listener channel is closed");
  }
  this.isa = address;
  this.sleeper = new Sleeper(1000, this);
}
 

import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的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.ipc.RpcServer.BlockingServiceAndInterface; //导入依赖的package包/类
/**
 * @return list of blocking services and their security info classes that this server supports
 */
private List<BlockingServiceAndInterface> getServices() {
  List<BlockingServiceAndInterface> bssi = new ArrayList<BlockingServiceAndInterface>(3);
  bssi.add(new BlockingServiceAndInterface(
      MasterProtos.MasterService.newReflectiveBlockingService(this),
      MasterProtos.MasterService.BlockingInterface.class));
  bssi.add(new BlockingServiceAndInterface(
      RegionServerStatusProtos.RegionServerStatusService.newReflectiveBlockingService(this),
      RegionServerStatusProtos.RegionServerStatusService.BlockingInterface.class));
  return bssi;
}
 

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