Java 类io.grpc.ServerMethodDefinition 实例源码

/** Never returns {@code null}. */
private <ReqT, RespT> ServerStreamListener startCall(ServerStream stream, String fullMethodName,
    ServerMethodDefinition<ReqT, RespT> methodDef, Metadata headers,
    Context.CancellableContext context, StatsTraceContext statsTraceCtx) {
  // TODO(ejona86): should we update fullMethodName to have the canonical path of the method?
  ServerCallImpl<ReqT, RespT> call = new ServerCallImpl<ReqT, RespT>(
      stream, methodDef.getMethodDescriptor(), headers, context,
      decompressorRegistry, compressorRegistry);
  ServerCallHandler<ReqT, RespT> callHandler = methodDef.getServerCallHandler();
  statsTraceCtx.serverCallStarted(
      new ServerCallInfoImpl<ReqT, RespT>(
          methodDef.getMethodDescriptor(),
          call.getAttributes(),
          call.getAuthority()));

  for (ServerInterceptor interceptor : interceptors) {
    callHandler = InternalServerInterceptors.interceptCallHandler(interceptor, callHandler);
  }
  ServerCall.Listener<ReqT> listener = callHandler.startCall(call, headers);
  if (listener == null) {
    throw new NullPointerException(
        "startCall() returned a null listener for method " + fullMethodName);
  }
  return call.newServerStreamListener(listener);
}

@Test
public void replaceAndLookup() {
  assertNull(registry.addService(basicServiceDefinition));
  assertNotNull(registry.lookupMethod("basic/flow"));
  MethodDescriptor<String, Integer> anotherMethod = MethodDescriptor.<String, Integer>newBuilder()
      .setType(MethodType.UNKNOWN)
      .setFullMethodName("basic/another")
      .setRequestMarshaller(requestMarshaller)
      .setResponseMarshaller(responseMarshaller)
      .build();
  ServerServiceDefinition replaceServiceDefinition = ServerServiceDefinition.builder(
      new ServiceDescriptor("basic", anotherMethod))
      .addMethod(anotherMethod, flowHandler).build();
  ServerMethodDefinition<?, ?> anotherMethodDefinition =
      replaceServiceDefinition.getMethod("basic/another");
  assertSame(basicServiceDefinition, registry.addService(replaceServiceDefinition));

  assertNull(registry.lookupMethod("basic/flow"));
  ServerMethodDefinition<?, ?> method = registry.lookupMethod("basic/another");
  assertSame(anotherMethodDefinition, method);
}

@Override
public ServerMethodDefinition<?, ?> lookupMethod(String methodName, @Nullable String authority) {
  return ServerMethodDefinition.create(
      MethodDescriptor.<byte[], byte[]>newBuilder()
          .setRequestMarshaller(new ByteArrayMarshaller())
          .setResponseMarshaller(new ByteArrayMarshaller())
          .setType(MethodType.UNARY)
          .setFullMethodName(methodName)
          .build(),
      ServerCalls.asyncUnaryCall(new ProxyUnaryMethod(backend, methodName)));
}

@Nullable
private <I, O> ArmeriaServerCall<I, O> startCall(
        String fullMethodName,
        ServerMethodDefinition<I, O> methodDef,
        ServiceRequestContext ctx,
        HttpHeaders headers,
        HttpResponseWriter res,
        SerializationFormat serializationFormat) {
    ArmeriaServerCall<I, O> call = new ArmeriaServerCall<>(
            headers,
            methodDef.getMethodDescriptor(),
            compressorRegistry,
            decompressorRegistry,
            res,
            maxInboundMessageSizeBytes,
            maxOutboundMessageSizeBytes,
            ctx,
            serializationFormat,
            jsonMarshaller);
    final ServerCall.Listener<I> listener;
    try (SafeCloseable ignored = RequestContext.push(ctx)) {
        listener = methodDef.getServerCallHandler().startCall(call, EMPTY_METADATA);
    } catch (Throwable t) {
        call.setListener(new EmptyListener<>());
        call.close(Status.fromThrowable(t), EMPTY_METADATA);
        logger.warn(
                "Exception thrown from streaming request stub method before processing any request data" +
                " - this is likely a bug in the stub implementation.");
        return null;
    }
    if (listener == null) {
        // This will never happen for normal generated stubs but could conceivably happen for manually
        // constructed ones.
        throw new NullPointerException(
                "startCall() returned a null listener for method " + fullMethodName);
    }
    call.setListener(listener);
    return call;
}

private static MessageMarshaller jsonMarshaller(HandlerRegistry registry) {
    List<MethodDescriptor<?, ?>> methods =
            registry.services().stream()
                    .flatMap(service -> service.getMethods().stream())
                    .map(ServerMethodDefinition::getMethodDescriptor)
                    .collect(toImmutableList());
    return GrpcJsonUtil.jsonMarshaller(methods);
}

HandlerRegistry build() {
    ImmutableMap.Builder<String, ServerMethodDefinition<?, ?>> mapBuilder =
            ImmutableMap.builder();
    for (ServerServiceDefinition service : services.values()) {
        for (ServerMethodDefinition<?, ?> method : service.getMethods()) {
            mapBuilder.put(method.getMethodDescriptor().getFullMethodName(), method);
        }
    }
    return new HandlerRegistry(ImmutableList.copyOf(services.values()), mapBuilder.build());
}

/**
 * Creates a new instance that decorates the specified {@link Service}.
 */
UnframedGrpcService(Service<HttpRequest, HttpResponse> delegate) {
    super(delegate);
    delegateGrpcService =
            delegate.as(GrpcService.class)
                    .orElseThrow(
                            () -> new IllegalArgumentException("Decorated service must be a GrpcService."));
    methodsByName = delegateGrpcService.services()
                               .stream()
                               .flatMap(service -> service.getMethods().stream())
                               .map(ServerMethodDefinition::getMethodDescriptor)
                               .collect(ImmutableMap.toImmutableMap(MethodDescriptor::getFullMethodName,
                                                                    Function.identity()));
}

InternalHandlerRegistry build() {
  Map<String, ServerMethodDefinition<?, ?>> map =
      new HashMap<String, ServerMethodDefinition<?, ?>>();
  for (ServerServiceDefinition service : services.values()) {
    for (ServerMethodDefinition<?, ?> method : service.getMethods()) {
      map.put(method.getMethodDescriptor().getFullMethodName(), method);
    }
  }
  return new InternalHandlerRegistry(
      Collections.unmodifiableList(new ArrayList<ServerServiceDefinition>(services.values())),
      Collections.unmodifiableMap(map));
}

/**
 * Note: This does not actually honor the authority provided.  It will, eventually in the future.
 */
@Override
@Nullable
public ServerMethodDefinition<?, ?> lookupMethod(String methodName, @Nullable String authority) {
  String serviceName = MethodDescriptor.extractFullServiceName(methodName);
  if (serviceName == null) {
    return null;
  }
  ServerServiceDefinition service = services.get(serviceName);
  if (service == null) {
    return null;
  }
  return service.getMethod(methodName);
}

@Test
public void simpleLookup() {
  assertNull(registry.addService(basicServiceDefinition));
  ServerMethodDefinition<?, ?> method = registry.lookupMethod("basic/flow");
  assertSame(flowMethodDefinition, method);

  assertNull(registry.lookupMethod("/basic/flow"));
  assertNull(registry.lookupMethod("basic/basic"));
  assertNull(registry.lookupMethod("flow/flow"));
  assertNull(registry.lookupMethod("completely/random"));
}

@Test
public void simpleLookupWithBindable() {
  BindableService bindableService =
      new BindableService() {
        @Override
        public ServerServiceDefinition bindService() {
          return basicServiceDefinition;
        }
      };

  assertNull(registry.addService(bindableService));

  ServerMethodDefinition<?, ?> method = registry.lookupMethod("basic/flow");
  assertSame(flowMethodDefinition, method);
}

@SuppressWarnings("unchecked")
private static ServerMethodDefinition<Void, Void> getSoleMethod(
    ServerServiceDefinition serviceDef) {
  if (serviceDef.getMethods().size() != 1) {
    throw new AssertionError("Not exactly one method present");
  }
  return (ServerMethodDefinition<Void, Void>) getOnlyElement(serviceDef.getMethods());
}

@Override
public ServerMethodDefinition<?, ?> lookupMethod(String methodName, String authority) {
  // TODO Auto-generated method stub
  return null;
}

@Override
protected HttpResponse doPost(ServiceRequestContext ctx, HttpRequest req) throws Exception {
    MediaType contentType = req.headers().contentType();
    SerializationFormat serializationFormat = findSerializationFormat(contentType);
    if (serializationFormat == null) {
        return HttpResponse.of(HttpStatus.UNSUPPORTED_MEDIA_TYPE,
                               MediaType.PLAIN_TEXT_UTF_8,
                               "Missing or invalid Content-Type header.");
    }

    ctx.logBuilder().serializationFormat(serializationFormat);

    String methodName = GrpcRequestUtil.determineMethod(ctx);
    if (methodName == null) {
        return HttpResponse.of(HttpStatus.BAD_REQUEST,
                               MediaType.PLAIN_TEXT_UTF_8,
                               "Invalid path.");
    }

    ServerMethodDefinition<?, ?> method = registry.lookupMethod(methodName);
    if (method == null) {
        return HttpResponse.of(
                ArmeriaServerCall.statusToTrailers(
                        Status.UNIMPLEMENTED.withDescription("Method not found: " + methodName),
                        false));
    }

    ctx.logBuilder().requestContent(GrpcLogUtil.rpcRequest(method.getMethodDescriptor()), null);

    String timeoutHeader = req.headers().get(GrpcHeaderNames.GRPC_TIMEOUT);
    if (timeoutHeader != null) {
        try {
            long timeout = TimeoutHeaderUtil.fromHeaderValue(timeoutHeader);
            ctx.setRequestTimeout(Duration.ofNanos(timeout));
        } catch (IllegalArgumentException e) {
            return HttpResponse.of(ArmeriaServerCall.statusToTrailers(Status.fromThrowable(e), false));
        }
    }

    HttpResponseWriter res = HttpResponse.streaming();
    ArmeriaServerCall<?, ?> call = startCall(
            methodName, method, ctx, req.headers(), res, serializationFormat);
    if (call != null) {
        ctx.setRequestTimeoutHandler(() -> {
            call.close(Status.DEADLINE_EXCEEDED, EMPTY_METADATA);
        });
        req.subscribe(call.messageReader(), ctx.eventLoop(), true);
    }
    return res;
}

private HandlerRegistry(List<ServerServiceDefinition> services,
                        Map<String, ServerMethodDefinition<?, ?>> methods) {
    this.services = requireNonNull(services, "services");
    this.methods = requireNonNull(methods, "methods");
}

@Nullable
ServerMethodDefinition<?, ?> lookupMethod(String methodName) {
    return methods.get(methodName);
}

Map<String, ServerMethodDefinition<?, ?>> methods() {
    return methods;
}

private InternalHandlerRegistry(
    List<ServerServiceDefinition> services, Map<String, ServerMethodDefinition<?, ?>> methods) {
  this.services = services;
  this.methods = methods;
}

@Nullable
ServerMethodDefinition<?, ?> lookupMethod(String methodName) {
  return methods.get(methodName);
}

@Override
public ServerMethodDefinition<?, ?> lookupMethod(String methodName,
    @Nullable String authority) {
  return null;
}

@Override
public ServerMethodDefinition<?, ?> lookupMethod(
    String methodName, @Nullable String authority) {
  return mutableFallbackRegistry.lookupMethod(methodName, authority);
}