Java 类io.grpc.stub.AbstractStub 实例源码

@SuppressWarnings({"rawtypes", "unchecked"})
private GrpcProtocolClient<Object> buildProtoClient(GrpcURL refUrl) {
  boolean isGeneric = refUrl.getParameter(Constants.GENERIC_KEY, Boolean.FALSE);
  boolean isGrpcStub = refUrl.getParameter(Constants.GRPC_STUB_KEY, Boolean.FALSE);
  if (isGeneric) {
    return new GenericProxyClient<Object>(refUrl);
  } else {
    if (isGrpcStub) {
      String stubClassName = refUrl.getParameter(Constants.INTERFACECLASS_KEY);
      try {
        Class<? extends AbstractStub> stubClass =
            (Class<? extends AbstractStub>) ReflectUtils.name2class(stubClassName);
        return new GrpcStubClient<Object>(stubClass, refUrl);
      } catch (ClassNotFoundException e) {
        throw new IllegalArgumentException("grpc stub client the class must exist in classpath",
            e);
      }
    } else {
      return new DefaultProxyClient<Object>(refUrl);
    }
  }
}

<T extends AbstractStub<T>, R> CompletableFuture<R> withNewChannel(
    String endpoint,
    Function<ManagedChannel, T> stubCustomizer,
    Function<T, CompletableFuture<R>> stubConsumer) {

  final ManagedChannel channel = defaultChannelBuilder()
      .nameResolverFactory(
          forEndpoints(
              Optional.ofNullable(builder.authority()).orElse("etcd"),
              Collections.singleton(endpoint),
              Optional.ofNullable(builder.uriResolverLoader())
                  .orElseGet(URIResolverLoader::defaultLoader)
          )
      ).build();

  try {
    T stub = stubCustomizer.apply(channel);

    return stubConsumer.apply(stub).whenComplete(
        (r, t) -> channel.shutdown()
    );
  } catch (Exception e) {
    channel.shutdown();
    throw EtcdExceptionFactory.toEtcdException(e);
  }
}

@SuppressWarnings("unchecked")
@Override
public void addConnection(ServiceType serviceType, String host, int port) {
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port)
            .negotiationType(NegotiationType.PLAINTEXT) // TODO: gRPC encryption
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)
            .directExecutor()
            .channelType(EpollSocketChannel.class)
            .eventLoopGroup(new EpollEventLoopGroup())
            .build();

    AbstractStub blocking;
    AbstractStub async;

    switch (serviceType) {
        case WORLD: {
            blocking = WorldServiceGrpc.newBlockingStub(channel);
            async = WorldServiceGrpc.newStub(channel);
            break;
        }
        case PLUGIN_MANAGER: {
            blocking = PluginManagerGrpc.newBlockingStub(channel);
            async = PluginManagerGrpc.newStub(channel);
            break;
        }
        default: {
            throw new RuntimeException("Service type not handled: " + serviceType.name());
        }
    }

    ServiceConnection connection = ServiceConnection.builder()
            .channel(channel)
            .blockingStub(blocking)
            .asyncStub(async)
            .build();

    this.connections.put(serviceType, connection);
}

private boolean isGrpcStubClient(Class<?> referenceClass) {
  if (AbstractStub.class.isAssignableFrom(referenceClass)) {
    return true;
  } else {
    return false;
  }
}

/**
 * Get remoting service instance for client invocation.
 * @param <T> T
 * @param type the type of expected service instance
 * @return the instance of AbstractStub.
 */
public <T extends AbstractStub<T>> T getRemotingService(Class<T> type) {
  checkArgument(type != null, "Param cannot be null!");
  checkState(channel != null, "Channel has not been initialized.");
  T service = null;
  try {
    Constructor<T> constructor = type.getDeclaredConstructor(Channel.class);
    constructor.setAccessible(true);
    service = constructor.newInstance(getChannel()).withDeadlineAfter(defaultTimeout, TimeUnit.MILLISECONDS);     
  } catch (Exception ex) {
    throw new RpcClientException("Get remoting service '" + type.getName() + "' error happend", ex);
  }
  return service;
}

@Override
public <T> Optional<ClientBuilderParams> clientBuilderParams(T client) {
    if (!(client instanceof AbstractStub)) {
        return Optional.empty();
    }
    AbstractStub<?> stub = (AbstractStub<?>) client;
    if (!(stub.getChannel() instanceof ArmeriaChannel)) {
        return Optional.empty();
    }
    return Optional.of((ClientBuilderParams) stub.getChannel());
}

/**
 * Get remoting service instance for client invocation.
 * 
 * @param type the type of expected service instance
 * @param timeout (millisecond) specify the remoting call will be expired at the specified offset 
 * @return T the instance of T.
 */
@Override
public AbstractStub<?> getRemotingService(Class<? extends AbstractStub<?>> type, String preferredZone, boolean usedTls, int timeout) {
  checkArgument(type != null, "The type of service interface cannot be null!");
  checkState(state.get() == State.STARTED, "Grpc client is not started!");
  AbstractStub<?> service;
  Method stubInitializationMethod;
  try {
    //The basic idea of instantiating grpc client stub:
    //  Class stubClazz = ***Stub.class (parameter pass in - Class<? extends AbstractStub<?>> type);
    //  Class grpcClazz = stubClazz.getEnclosingClass();
    //  Field serviceNameFiled = grpcClazz.getDeclaredField("SERVICE_NAME");
    //  String value = (String) serviceNameFiled.get(null);

    //E.g. public static final String SERVICE_NAME = "routeguide.RouteGuide";
    String serviceName = type.getEnclosingClass().getDeclaredField(GrpcConstants.SERVICE_NAME).get(null).toString();      
    if (!STUBS.containsKey(type.getName())) {
      // Instantiate the generated gRPC class.
      if (type.getName().endsWith(GrpcConstants.CLIENT_STUB_SUFFIX_BLOCKING)) {
        stubInitializationMethod = type.getEnclosingClass().getMethod(GrpcConstants.NEW_CLIENT_BLOCKING_STUB, Channel.class);
      } else if (type.getName().endsWith(GrpcConstants.CLIENT_STUB_SUFFIX_FUTURE)) {
        stubInitializationMethod = type.getEnclosingClass().getMethod(GrpcConstants.NEW_CLIENT_FUTURE_STUB, Channel.class);
      } else {
        stubInitializationMethod = type.getEnclosingClass().getMethod(GrpcConstants.NEW_CLIENT_ASYN_STUB, Channel.class);
      }
      STUBS.putIfAbsent(type.getName(), stubInitializationMethod);
    } else {
      stubInitializationMethod = STUBS.get(type.getName());
    }
    // instantiate the client stub according to the stub type
    service = (AbstractStub<?>) stubInitializationMethod.invoke(null, initializer.getChannelPool().getChannel(serviceName, preferredZone, usedTls));
    //Customizes the CallOptions passed the deadline to interceptor
    if (timeout > 0) {
       service.withOption(StubDeadlineClientInterceptor.DEADLINE_KEY, Integer.valueOf(timeout));
    }

  } catch (Exception ex) {
    log.error("Get remoting service '{}' error happend", type.getName(), ex);
    throw new RpcClientException(ex);
  }
  return service;
}

@Override
public RpcClient<? extends AbstractStub<?>> getRpcClient() {

  return GrpcClient.getInstance();
}

private static <T extends AbstractStub<T>> T withCallCredentials(
    T stub, @Nullable CallCredentials callCredentials) {
  stub = callCredentials != null ? stub.withCallCredentials(callCredentials) : stub;
  return stub;
}

/**
 * create stub with saved channel.
 *
 * @param supplier the stub supplier
 * @param <T> the type of stub
 * @return the attached stub
 */
<T extends AbstractStub<T>> T newStub(Function<ManagedChannel, T> supplier) {
  return supplier.apply(getChannel());
}

/**
 * Get remoting service instance for client invocation.
 * @param <T> T
 * @param type the type of expected service instance
 * @return the instance of AbstractStub.
 */
public <T extends AbstractStub<T>> T getRemotingService(Class<T> type) {

  return ServiceTestResource.client.getRemotingService(type);
}