Java 类com.google.protobuf.RpcChannel 实例源码

private <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, RegionInfo region, byte[] row) {
  RegionCoprocessorRpcChannelImpl channel = new RegionCoprocessorRpcChannelImpl(conn, tableName,
      region, row, rpcTimeoutNs, operationTimeoutNs);
  S stub = stubMaker.apply(channel);
  CompletableFuture<R> future = new CompletableFuture<>();
  ClientCoprocessorRpcController controller = new ClientCoprocessorRpcController();
  callable.call(stub, controller, resp -> {
    if (controller.failed()) {
      future.completeExceptionally(controller.getFailed());
    } else {
      future.complete(resp);
    }
  });
  return future;
}

@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable) {
  MasterCoprocessorRpcChannelImpl channel =
      new MasterCoprocessorRpcChannelImpl(this.<Message> newMasterCaller());
  S stub = stubMaker.apply(channel);
  CompletableFuture<R> future = new CompletableFuture<>();
  ClientCoprocessorRpcController controller = new ClientCoprocessorRpcController();
  callable.call(stub, controller, resp -> {
    if (controller.failed()) {
      future.completeExceptionally(controller.getFailed());
    } else {
      future.complete(resp);
    }
  });
  return future;
}

@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, ServerName serverName) {
  RegionServerCoprocessorRpcChannelImpl channel =
      new RegionServerCoprocessorRpcChannelImpl(this.<Message> newServerCaller().serverName(
        serverName));
  S stub = stubMaker.apply(channel);
  CompletableFuture<R> future = new CompletableFuture<>();
  ClientCoprocessorRpcController controller = new ClientCoprocessorRpcController();
  callable.call(stub, controller, resp -> {
    if (controller.failed()) {
      future.completeExceptionally(controller.getFailed());
    } else {
      future.complete(resp);
    }
  });
  return future;
}

@Test
public void testAsyncConnectionSetup() throws Exception {
  TestRpcServer rpcServer = new TestRpcServer();
  AsyncRpcClient client = createRpcClient(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);

    channel.callMethod(md, new PayloadCarryingRpcController(), param, md.getOutputType()
        .toProto(), new RpcCallback<Message>() {
      @Override
      public void run(Message parameter) {
        done.set(true);
      }
    });

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

private void doTest(RpcServer rpcServer) throws InterruptedException,
    ServiceException, IOException {
  BlockingRpcChannel blockingChannel = RpcChannels
      .newBlockingRpcChannel(clientConnectionFactory);
  RpcChannel channel = RpcChannels.newRpcChannel(clientConnectionFactory,
      threadPool);
  BlockingInterface blockingStub = TestService
      .newBlockingStub(blockingChannel);
  TestService stub = TestService.newStub(channel);

  try {
    rpcServer.startServer();
    Thread.sleep(500);

    doRpc(stub);
    doBlockingRpc(blockingStub);
    doBlockingRpc(blockingStub);
    doRpc(stub);
  } finally {
    Thread.sleep(500);
    System.out.println("Closing Client");
    if (clientConnectionFactory instanceof Closeable) {
      ((PersistentRpcConnectionFactory) clientConnectionFactory).close();
    }
    Thread.sleep(100);
    System.out.println("Closing Server");
    rpcServer.shutDown();
  }
}

private <S, R> void onLocateComplete(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, CoprocessorCallback<R> callback,
    List<HRegionLocation> locs, byte[] endKey, boolean endKeyInclusive,
    AtomicBoolean locateFinished, AtomicInteger unfinishedRequest, HRegionLocation loc,
    Throwable error) {
  if (error != null) {
    callback.onError(error);
    return;
  }
  unfinishedRequest.incrementAndGet();
  RegionInfo region = loc.getRegion();
  if (locateFinished(region, endKey, endKeyInclusive)) {
    locateFinished.set(true);
  } else {
    conn.getLocator()
        .getRegionLocation(tableName, region.getEndKey(), RegionLocateType.CURRENT,
          operationTimeoutNs)
        .whenComplete((l, e) -> onLocateComplete(stubMaker, callable, callback, locs, endKey,
          endKeyInclusive, locateFinished, unfinishedRequest, l, e));
  }
  coprocessorService(stubMaker, callable, region, region.getStartKey()).whenComplete((r, e) -> {
    if (e != null) {
      callback.onRegionError(region, e);
    } else {
      callback.onRegionComplete(region, r);
    }
    if (unfinishedRequest.decrementAndGet() == 0 && locateFinished.get()) {
      callback.onComplete();
    }
  });
}

private AddressBookProtos.AddressBookService connectExportedProtobufAddress(ExportDeclaration declaration) throws EndpointException, NoSuchMethodException, InvocationTargetException, IllegalAccessException, BinderException {
    ProtobufferExportDeclarationWrapper pojo = ProtobufferExportDeclarationWrapper.create(declaration);
    Bus cxfbus = BusFactory.getThreadDefaultBus();
    BindingFactoryManager mgr = cxfbus.getExtension(BindingFactoryManager.class);
    mgr.registerBindingFactory(ProtobufBindingFactory.PROTOBUF_BINDING_ID, new ProtobufBindingFactory(cxfbus));
    Class<?> bufferService = AddressBookProtos.AddressBookService.class;
    Class<?> bufferMessage = AddressBookProtos.AddressBookServiceMessage.class;
    Class<? extends Message> generic = bufferMessage.asSubclass(Message.class);
    RpcChannel channel = new SimpleRpcChannel(pojo.getAddress(), generic);
    Method method = bufferService.getMethod("newStub", RpcChannel.class);
    Object service = method.invoke(bufferService, channel);
    AddressBookProtos.AddressBookService addressBook = (AddressBookProtos.AddressBookService) service;
    return addressBook;
}

@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();
  }
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, byte[] row) {
  return coprocessorService(stubMaker, callable, null, row);
}

public CoprocessorServiceBuilderImpl(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, CoprocessorCallback<R> callback) {
  this.stubMaker = Preconditions.checkNotNull(stubMaker, "stubMaker is null");
  this.callable = Preconditions.checkNotNull(callable, "callable is null");
  this.callback = Preconditions.checkNotNull(callback, "callback is null");
}

@Override
public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
    Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
    CoprocessorCallback<R> callback) {
  return new CoprocessorServiceBuilderImpl<>(stubMaker, callable, callback);
}

@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable) {
  return wrap(rawAdmin.coprocessorService(stubMaker, callable));
}

@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, ServerName serverName) {
  return wrap(rawAdmin.coprocessorService(stubMaker, callable, serverName));
}

@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, byte[] row) {
  return wrap(rawTable.coprocessorService(stubMaker, callable, row));
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

@SuppressWarnings("unchecked")
public static <T extends Service> T newServiceStub(Class<T> service, RpcChannel channel)
    throws Exception {
  return (T)Methods.call(service, null, "newStub",
      new Class[]{ RpcChannel.class }, new Object[]{ channel });
}

/**
 * Creates a "channel" that can be used by a protobuf service.  Useful setting up
 * protobuf stubs.
 *
 * @param sn server name describing location of server
 * @param user which is to use the connection
 * @param rpcTimeout default rpc operation timeout
 *
 * @return A rpc channel that goes via this rpc client instance.
 */
public RpcChannel createRpcChannel(final ServerName sn, final User user, int rpcTimeout) {
  return new RpcChannelImplementation(this, sn, user, rpcTimeout);
}

/**
 * Create a {@link RpcChannel} that uses the given
 * {@link RpcConnectionFactory} to connect to the RPC server and the given
 * {@link Executor} to listen for the RPC response after sending the request.
 * RPCs made using this {@link RpcChannel} will not block the thread calling
 * the RPC method. Use {@link #newBlockingRpcChannel(RpcConnectionFactory)} if
 * you want the RPC method to block.
 * <p>
 * This channel doesn't call the callback if the server-side implementation
 * did not call the callback. If any error occurs, it will call the callback
 * with null and update the controller with the error.
 */
public static RpcChannel newRpcChannel(
    RpcConnectionFactory connectionFactory, Executor executor) {
  return new RpcChannelImpl(connectionFactory, executor);
}

/**
 * Execute the given coprocessor call on the region which contains the given {@code row}.
 * <p>
 * The {@code stubMaker} is just a delegation to the {@code newStub} call. Usually it is only a
 * one line lambda expression, like:
 *
 * <pre>
 * <code>
 * channel -> xxxService.newStub(channel)
 * </code>
 * </pre>
 *
 * @param stubMaker a delegation to the actual {@code newStub} call.
 * @param callable a delegation to the actual protobuf rpc call. See the comment of
 *          {@link ServiceCaller} for more details.
 * @param row The row key used to identify the remote region location
 * @param <S> the type of the asynchronous stub
 * @param <R> the type of the return value
 * @return the return value of the protobuf rpc call, wrapped by a {@link CompletableFuture}.
 * @see ServiceCaller
 */
<S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, byte[] row);

/**
 * Execute a coprocessor call on the regions which are covered by a range.
 * <p>
 * Use the returned {@link CoprocessorServiceBuilder} construct your request and then execute it.
 * <p>
 * The {@code stubMaker} is just a delegation to the {@code xxxService.newStub} call. Usually it
 * is only a one line lambda expression, like:
 *
 * <pre>
 * <code>
 * channel -> xxxService.newStub(channel)
 * </code>
 * </pre>
 *
 * @param stubMaker a delegation to the actual {@code newStub} call.
 * @param callable a delegation to the actual protobuf rpc call. See the comment of
 *          {@link ServiceCaller} for more details.
 * @param callback callback to get the response. See the comment of {@link CoprocessorCallback}
 *          for more details.
 */
<S, R> CoprocessorServiceBuilder<S, R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable, CoprocessorCallback<R> callback);

/**
 * Execute the given coprocessor call on the master.
 * <p>
 * The {@code stubMaker} is just a delegation to the {@code newStub} call. Usually it is only a
 * one line lambda expression, like:
 *
 * <pre>
 * <code>
 * channel -> xxxService.newStub(channel)
 * </code>
 * </pre>
 * @param stubMaker a delegation to the actual {@code newStub} call.
 * @param callable a delegation to the actual protobuf rpc call. See the comment of
 *          {@link ServiceCaller} for more details.
 * @param <S> the type of the asynchronous stub
 * @param <R> the type of the return value
 * @return the return value of the protobuf rpc call, wrapped by a {@link CompletableFuture}.
 * @see ServiceCaller
 */
<S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
    ServiceCaller<S, R> callable);

/**
 * Execute the given coprocessor call on the given region server.
 * <p>
 * The {@code stubMaker} is just a delegation to the {@code newStub} call. Usually it is only a
 * one line lambda expression, like:
 *
 * <pre>
 * <code>
 * channel -> xxxService.newStub(channel)
 * </code>
 * </pre>
 * @param stubMaker a delegation to the actual {@code newStub} call.
 * @param callable a delegation to the actual protobuf rpc call. See the comment of
 *          {@link ServiceCaller} for more details.
 * @param serverName the given region server
 * @param <S> the type of the asynchronous stub
 * @param <R> the type of the return value
 * @return the return value of the protobuf rpc call, wrapped by a {@link CompletableFuture}.
 * @see ServiceCaller
 */
<S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
  ServiceCaller<S, R> callable, ServerName serverName);

/**
 * Creates a "channel" that can be used by a protobuf service.  Useful setting up
 * protobuf stubs.
 *
 * @param sn server name describing location of server
 * @param user which is to use the connection
 * @param rpcTimeout default rpc operation timeout
 *
 * @return A rpc channel that goes via this rpc client instance.
 * @throws IOException when channel could not be created
 */
public RpcChannel createRpcChannel(final ServerName sn, final User user, int rpcTimeout) {
  return new RpcChannelImplementation(this, sn, user, rpcTimeout);
}

/**
 * Creates and returns a {@link com.google.protobuf.RpcChannel} instance connected to the
 * table region containing the specified row.  The row given does not actually have
 * to exist.  Whichever region would contain the row based on start and end keys will
 * be used.  Note that the {@code row} parameter is also not passed to the
 * coprocessor handler registered for this protocol, unless the {@code row}
 * is separately passed as an argument in the service request.  The parameter
 * here is only used to locate the region used to handle the call.
 * <p/>
 * <p>
 * The obtained {@link com.google.protobuf.RpcChannel} instance can be used to access a published
 * coprocessor {@link com.google.protobuf.Service} using standard protobuf service invocations:
 * </p>
 * <p/>
 * <div style="background-color: #cccccc; padding: 2px">
 * <blockquote><pre>
 * CoprocessorRpcChannel channel = myTable.coprocessorService(rowkey);
 * MyService.BlockingInterface service = MyService.newBlockingStub(channel);
 * MyCallRequest request = MyCallRequest.newBuilder()
 *     ...
 *     .build();
 * MyCallResponse response = service.myCall(null, request);
 * </pre></blockquote></div>
 *
 * @param table to get service from
 * @param row   The row key used to identify the remote region location
 * @return A CoprocessorRpcChannel instance
 * @throws java.io.IOException when there was an error creating connection or getting location
 */
public RpcChannel coprocessorService(TableName table, byte[] row) throws IOException {
  HRegionLocation location = getRegionLocation(table, row, false);

  return client.createRpcChannel(location.getServerName(), User.getCurrent(), rpcTimeout);
}