TestAsyncIPC.java 文件源码

@Test
public void testRTEDuringAsyncConnectionSetup() throws Exception {
  TestRpcServer rpcServer = new TestRpcServer();
  AsyncRpcClient client = createRpcClientRTEDuringConnectionSetup(CONF);
  try {
    rpcServer.start();
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
    EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();

    RpcChannel channel =
        client.createRpcChannel(ServerName.valueOf(address.getHostName(), address.getPort(),
          System.currentTimeMillis()), User.getCurrent(), 0);

    final AtomicBoolean done = new AtomicBoolean(false);

    PayloadCarryingRpcController controller = new PayloadCarryingRpcController();
    controller.notifyOnFail(new RpcCallback<IOException>() {
      @Override
      public void run(IOException e) {
        done.set(true);
        LOG.info("Caught expected exception: " + e.toString());
        assertTrue(StringUtils.stringifyException(e).contains("Injected fault"));
      }
    });

    channel.callMethod(md, controller, param, md.getOutputType().toProto(),
      new RpcCallback<Message>() {
        @Override
        public void run(Message parameter) {
          done.set(true);
          fail("Expected an exception to have been thrown!");
        }
      });

    TEST_UTIL.waitFor(1000, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        return done.get();
      }
    });
  } finally {
    client.close();
    rpcServer.stop();
  }
}