Java 类io.grpc.MethodDescriptor.MethodType 实例源码

@GuardedBy("lock")
private void startStream(OkHttpClientStream stream) {
  Preconditions.checkState(
      stream.id() == OkHttpClientStream.ABSENT_ID, "StreamId already assigned");
  streams.put(nextStreamId, stream);
  setInUse();
  stream.transportState().start(nextStreamId);
  // For unary and server streaming, there will be a data frame soon, no need to flush the header.
  if ((stream.getType() != MethodType.UNARY && stream.getType() != MethodType.SERVER_STREAMING)
      || stream.useGet()) {
    frameWriter.flush();
  }
  if (nextStreamId >= Integer.MAX_VALUE - 2) {
    // Make sure nextStreamId greater than all used id, so that mayHaveCreatedStream() performs
    // correctly.
    nextStreamId = Integer.MAX_VALUE;
    startGoAway(Integer.MAX_VALUE, ErrorCode.NO_ERROR,
        Status.UNAVAILABLE.withDescription("Stream ids exhausted"));
  } else {
    nextStreamId += 2;
  }
}

@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  methodDescriptor = MethodDescriptor.<Void, Void>newBuilder()
      .setType(MethodDescriptor.MethodType.UNARY)
      .setFullMethodName("testService/test")
      .setRequestMarshaller(marshaller)
      .setResponseMarshaller(marshaller)
      .build();

  stream = new OkHttpClientStream(
      methodDescriptor,
      new Metadata(),
      frameWriter,
      transport,
      flowController,
      lock,
      MAX_MESSAGE_SIZE,
      "localhost",
      "userAgent",
      StatsTraceContext.NOOP,
      transportTracer);
}

ClientCallImpl(
    MethodDescriptor<ReqT, RespT> method, Executor executor, CallOptions callOptions,
    ClientTransportProvider clientTransportProvider,
    ScheduledExecutorService deadlineCancellationExecutor,
    CallTracer channelCallsTracer) {
  this.method = method;
  // If we know that the executor is a direct executor, we don't need to wrap it with a
  // SerializingExecutor. This is purely for performance reasons.
  // See https://github.com/grpc/grpc-java/issues/368
  this.callExecutor = executor == directExecutor()
      ? new SerializeReentrantCallsDirectExecutor()
      : new SerializingExecutor(executor);
  this.channelCallsTracer = channelCallsTracer;
  // Propagate the context from the thread which initiated the call to all callbacks.
  this.context = Context.current();
  this.unaryRequest = method.getType() == MethodType.UNARY
      || method.getType() == MethodType.SERVER_STREAMING;
  this.callOptions = callOptions;
  this.clientTransportProvider = clientTransportProvider;
  this.deadlineCancellationExecutor = deadlineCancellationExecutor;
}

@Test
public void idempotent() {
  MethodDescriptor<String, String> descriptor = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.SERVER_STREAMING)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(new StringMarshaller())
      .setResponseMarshaller(new StringMarshaller())
      .build();

  assertFalse(descriptor.isIdempotent());

  // Create a new desriptor by setting idempotent to true
  MethodDescriptor<String, String> newDescriptor =
      descriptor.toBuilder().setIdempotent(true).build();
  assertTrue(newDescriptor.isIdempotent());
  // All other fields should staty the same
  assertEquals(MethodType.SERVER_STREAMING, newDescriptor.getType());
  assertEquals("package.service/method", newDescriptor.getFullMethodName());
}

@Test
public void safe() {
  MethodDescriptor<String, String> descriptor = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.UNARY)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(new StringMarshaller())
      .setResponseMarshaller(new StringMarshaller())
      .build();
  assertFalse(descriptor.isSafe());

  // Create a new desriptor by setting safe to true
  MethodDescriptor<String, String> newDescriptor = descriptor.toBuilder().setSafe(true).build();
  assertTrue(newDescriptor.isSafe());
  // All other fields should staty the same
  assertEquals(MethodType.UNARY, newDescriptor.getType());
  assertEquals("package.service/method", newDescriptor.getFullMethodName());
}

/** Set up for test. */
@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  flowMethod = MethodDescriptor.<String, Integer>newBuilder()
      .setType(MethodType.UNKNOWN)
      .setFullMethodName("basic/flow")
      .setRequestMarshaller(requestMarshaller)
      .setResponseMarshaller(responseMarshaller)
      .build();

  Mockito.when(handler.startCall(
      Mockito.<ServerCall<String, Integer>>any(), Mockito.<Metadata>any()))
          .thenReturn(listener);

  serviceDefinition = ServerServiceDefinition.builder(new ServiceDescriptor("basic", flowMethod))
      .addMethod(flowMethod, handler).build();
}

@Test
public void failsOnNonDuplicateNames() {
  @SuppressWarnings("deprecation") // MethodDescriptor.create
  List<MethodDescriptor<?, ?>> descriptors = Arrays.<MethodDescriptor<?, ?>>asList(
      MethodDescriptor.create(
          MethodType.UNARY,
          MethodDescriptor.generateFullMethodName("name", "method"),
          TestMethodDescriptors.voidMarshaller(),
          TestMethodDescriptors.voidMarshaller()),
      MethodDescriptor.create(
          MethodType.UNARY,
          MethodDescriptor.generateFullMethodName("name", "method"),
          TestMethodDescriptors.voidMarshaller(),
          TestMethodDescriptors.voidMarshaller()));

  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage("duplicate");

  new ServiceDescriptor("name", descriptors);
}

@Before
public void setUp() throws Exception {
  MockitoAnnotations.initMocks(this);
  MethodDescriptor<String, Integer> flowMethod = MethodDescriptor.<String, Integer>newBuilder()
      .setType(MethodType.UNKNOWN)
      .setFullMethodName("basic/flow")
      .setRequestMarshaller(requestMarshaller)
      .setResponseMarshaller(responseMarshaller)
      .build();
  basicServiceDefinition = ServerServiceDefinition.builder(
      new ServiceDescriptor("basic", flowMethod))
      .addMethod(flowMethod, flowHandler)
      .build();

  MethodDescriptor<String, Integer> coupleMethod =
      flowMethod.toBuilder().setFullMethodName("multi/couple").build();
  MethodDescriptor<String, Integer> fewMethod =
      flowMethod.toBuilder().setFullMethodName("multi/few").build();
  multiServiceDefinition = ServerServiceDefinition.builder(
      new ServiceDescriptor("multi", coupleMethod, fewMethod))
      .addMethod(coupleMethod, coupleHandler)
      .addMethod(fewMethod, fewHandler)
      .build();

  flowMethodDefinition = getOnlyElement(basicServiceDefinition.getMethods());
}

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

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  if (ReflectUtils.isToStringMethod(method)) {
    return AbstractClientInvocation.this.toString();
  } else {
    GrpcRequest request = this.buildGrpcRequest(method, args);
    requstValidator.doValidate(request);
    MethodType methodType = request.getMethodType();
    Channel channel = request.getChannel();
    try {
      switch (methodType) {
        case UNARY:
          return unaryCall(request, channel);
        case CLIENT_STREAMING:
          return streamCall(request, channel);
        case SERVER_STREAMING:
          return streamCall(request, channel);
        case BIDI_STREAMING:
          return streamCall(request, channel);
        default:
          RpcServiceException rpcFramwork =
              new RpcServiceException(RpcErrorMsgConstant.SERVICE_UNFOUND);
          throw rpcFramwork;
      }
    } finally {
      Object remote = GrpcCallOptions.getAffinity(request.getRefUrl())
          .get(GrpcCallOptions.GRPC_CURRENT_ADDR_KEY);
      log.debug(String.format("Service: %s  Method: %s  RemoteAddress: %s",
          request.getServiceName(), request.getMethodName(), String.valueOf(remote)));
    }
  }
}

/**
 * Makes an rpc to the remote endpoint and respects the supplied callback. Returns a future which
 * terminates once the call has ended. For calls which are single-request, this throws
 * {@link IllegalArgumentException} if the size of {@code requests} is not exactly 1.
 */
public ListenableFuture<Void> call(
    ImmutableList<DynamicMessage> requests,
    StreamObserver<DynamicMessage> responseObserver,
    CallOptions callOptions) {
  Preconditions.checkArgument(!requests.isEmpty(), "Can't make call without any requests");
  MethodType methodType = getMethodType();
  long numRequests = requests.size();
  if (methodType == MethodType.UNARY) {
    logger.info("Making unary call");
    Preconditions.checkArgument(numRequests == 1,
        "Need exactly 1 request for unary call, but got: " + numRequests);
    return callUnary(requests.get(0), responseObserver, callOptions);
  } else if (methodType == MethodType.SERVER_STREAMING) {
    logger.info("Making server streaming call");
    Preconditions.checkArgument(numRequests == 1,
        "Need exactly 1 request for server streaming call, but got: " + numRequests);
    return callServerStreaming(requests.get(0), responseObserver, callOptions);
  } else if (methodType == MethodType.CLIENT_STREAMING) {
    logger.info("Making client streaming call with " + requests.size() + " requests");
    return callClientStreaming(requests, responseObserver, callOptions);
  } else {
    // Bidi streaming.
    logger.info("Making bidi streaming call with " + requests.size() + " requests");
    return callBidiStreaming(requests, responseObserver, callOptions);
  }
}

/** Returns the appropriate method type based on whether the client or server expect streams. */
private MethodType getMethodType() {
  boolean clientStreaming = protoMethodDescriptor.toProto().getClientStreaming();
  boolean serverStreaming = protoMethodDescriptor.toProto().getServerStreaming();

  if (!clientStreaming && !serverStreaming) {
    return MethodType.UNARY;
  } else if (!clientStreaming && serverStreaming) {
    return MethodType.SERVER_STREAMING;
  } else if (clientStreaming && !serverStreaming) {
    return MethodType.CLIENT_STREAMING;
  } else {
    return MethodType.BIDI_STREAMING;
  }
}

@Override
public void messageRead(ByteBufOrStream message) {

    final I request;

    boolean success = false;
    try {
        // Special case for unary calls.
        if (messageReceived && method.getType() == MethodType.UNARY) {
            closeListener(Status.INTERNAL.withDescription(
                    "More than one request messages for unary call or server streaming call"));
            return;
        }
        messageReceived = true;

        if (isCancelled()) {
            return;
        }
        success = true;
    } finally {
        if (message.buf() != null && !success) {
            message.buf().release();
        }
    }

    try {
        request = marshaller.deserializeRequest(message);
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
    try (SafeCloseable ignored = RequestContext.push(ctx)) {
        listener.onMessage(request);
    } catch (Throwable t) {
        close(Status.fromThrowable(t), EMPTY_METADATA);
    }
}

@Test
public void getUnaryRequest() {
  MethodDescriptor<?, ?> getMethod = MethodDescriptor.<Void, Void>newBuilder()
      .setType(MethodDescriptor.MethodType.UNARY)
      .setFullMethodName("service/method")
      .setIdempotent(true)
      .setSafe(true)
      .setRequestMarshaller(marshaller)
      .setResponseMarshaller(marshaller)
      .build();
  stream = new OkHttpClientStream(getMethod, new Metadata(), frameWriter, transport,
      flowController, lock, MAX_MESSAGE_SIZE, "localhost", "good-application",
      StatsTraceContext.NOOP, transportTracer);
  stream.start(new BaseClientStreamListener());

  // GET streams send headers after halfClose is called.
  verify(frameWriter, times(0)).synStream(
      eq(false), eq(false), eq(3), eq(0), headersCaptor.capture());
  verify(transport, times(0)).streamReadyToStart(isA(OkHttpClientStream.class));

  byte[] msg = "request".getBytes(Charset.forName("UTF-8"));
  stream.writeMessage(new ByteArrayInputStream(msg));
  stream.halfClose();
  verify(transport).streamReadyToStart(eq(stream));
  stream.transportState().start(3);

  verify(frameWriter)
      .synStream(eq(true), eq(false), eq(3), eq(0), headersCaptor.capture());
  assertThat(headersCaptor.getValue()).contains(Headers.METHOD_GET_HEADER);
  assertThat(headersCaptor.getValue()).contains(
      new Header(Header.TARGET_PATH, "/" + getMethod.getFullMethodName() + "?"
          + BaseEncoding.base64().encode(msg)));
}

/**
 * Creates a new method descriptor that always creates zero length messages, and always parses to
 * null objects.
 *
 * @since 1.1.0
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/2600")
public static MethodDescriptor<Void, Void> voidMethod() {
  return MethodDescriptor.<Void, Void>newBuilder()
      .setType(MethodType.UNARY)
      .setFullMethodName(MethodDescriptor.generateFullMethodName("service_foo", "method_bar"))
      .setRequestMarshaller(TestMethodDescriptors.voidMarshaller())
      .setResponseMarshaller(TestMethodDescriptors.voidMarshaller())
      .build();
}

@Test
public void createMethodDescriptor() {
  @SuppressWarnings("deprecation") // MethodDescriptor.create
  MethodDescriptor<String, String> descriptor = MethodDescriptor.<String, String>create(
      MethodType.CLIENT_STREAMING, "package.service/method", new StringMarshaller(),
      new StringMarshaller());
  assertEquals(MethodType.CLIENT_STREAMING, descriptor.getType());
  assertEquals("package.service/method", descriptor.getFullMethodName());
  assertFalse(descriptor.isIdempotent());
  assertFalse(descriptor.isSafe());
}

@Test
public void safeAndNonUnary() {
  MethodDescriptor<String, String> descriptor = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.SERVER_STREAMING)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(new StringMarshaller())
      .setResponseMarshaller(new StringMarshaller())
      .build();

  thrown.expect(IllegalArgumentException.class);
  MethodDescriptor<String, String> unused = descriptor.toBuilder().setSafe(true).build();
}

@Test
public void sampledToLocalTracing() {
  MethodDescriptor<String, String> md1 = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.SERVER_STREAMING)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(new StringMarshaller())
      .setResponseMarshaller(new StringMarshaller())
      .setSampledToLocalTracing(true)
      .build();
  assertTrue(md1.isSampledToLocalTracing());

  MethodDescriptor<String, String> md2 = md1.toBuilder()
      .setFullMethodName("package.service/method2")
      .build();
  assertTrue(md2.isSampledToLocalTracing());

  // Same method name as md1, but not setting sampledToLocalTracing
  MethodDescriptor<String, String> md3 = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.SERVER_STREAMING)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(new StringMarshaller())
      .setResponseMarshaller(new StringMarshaller())
      .build();
  assertFalse(md3.isSampledToLocalTracing());

  MethodDescriptor<String, String> md4 = md3.toBuilder()
      .setFullMethodName("package.service/method2")
      .setSampledToLocalTracing(true)
      .build();
  assertTrue(md4.isSampledToLocalTracing());
}

@Test
public void toBuilderTest() {
  MethodDescriptor<String, String> md1 = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.UNARY)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(StringMarshaller.INSTANCE)
      .setResponseMarshaller(StringMarshaller.INSTANCE)
      .setSampledToLocalTracing(true)
      .setIdempotent(true)
      .setSafe(true)
      .setSchemaDescriptor(new Object())
      .build();
  // Verify that we are not using any default builder values, so if md1 and md2 matches,
  // it's because toBuilder explicitly copied it.
  MethodDescriptor<String, String> defaults = MethodDescriptor.<String, String>newBuilder()
      .setType(MethodType.UNARY)
      .setFullMethodName("package.service/method")
      .setRequestMarshaller(StringMarshaller.INSTANCE)
      .setResponseMarshaller(StringMarshaller.INSTANCE)
      .build();
  assertNotEquals(md1.isSampledToLocalTracing(), defaults.isSampledToLocalTracing());
  assertNotEquals(md1.isIdempotent(), defaults.isIdempotent());
  assertNotEquals(md1.isSafe(), defaults.isSafe());
  assertNotEquals(md1.getSchemaDescriptor(), defaults.getSchemaDescriptor());

  // Verify that the builder correctly copied over the values
  MethodDescriptor<Integer, Integer> md2 = md1.toBuilder(
      IntegerMarshaller.INSTANCE,
      IntegerMarshaller.INSTANCE).build();
  assertSame(md1.getType(), md2.getType());
  assertSame(md1.getFullMethodName(), md2.getFullMethodName());
  assertSame(IntegerMarshaller.INSTANCE, md2.getRequestMarshaller());
  assertSame(IntegerMarshaller.INSTANCE, md2.getResponseMarshaller());
  assertEquals(md1.isSampledToLocalTracing(), md2.isSampledToLocalTracing());
  assertEquals(md1.isIdempotent(), md2.isIdempotent());
  assertEquals(md1.isSafe(), md2.isSafe());
  assertSame(md1.getSchemaDescriptor(), md2.getSchemaDescriptor());
}

@Test
public void failsOnNonMatchingNames() {
  @SuppressWarnings("deprecation") // MethodDescriptor.create
  List<MethodDescriptor<?, ?>> descriptors = Collections.<MethodDescriptor<?, ?>>singletonList(
      MethodDescriptor.create(
          MethodType.UNARY,
          MethodDescriptor.generateFullMethodName("wrongservice", "method"),
          TestMethodDescriptors.voidMarshaller(),
          TestMethodDescriptors.voidMarshaller()));

  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage("service names");

  new ServiceDescriptor("name", descriptors);
}

private GrpcMethod(String serviceName, String methodName, MethodType type) {
  this.serviceName = serviceName;
  this.methodName = methodName;
  this.type = type;
}

boolean streamsRequests() {
  return type == MethodType.CLIENT_STREAMING || type == MethodType.BIDI_STREAMING;
}

boolean streamsResponses() {
  return type == MethodType.SERVER_STREAMING || type == MethodType.BIDI_STREAMING;
}

@Override
public HttpResponse serve(ServiceRequestContext ctx, HttpRequest req) throws Exception {
    final HttpHeaders clientHeaders = req.headers();
    final MediaType contentType = clientHeaders.contentType();
    if (contentType == null) {
        // All gRPC requests, whether framed or non-framed, must have content-type. If it's not sent, let
        // the delegate return its usual error message.
        return delegate().serve(ctx, req);
    }

    for (SerializationFormat format : GrpcSerializationFormats.values()) {
        if (format.isAccepted(contentType)) {
            // Framed request, so just delegate.
            return delegate().serve(ctx, req);
        }
    }

    String methodName = GrpcRequestUtil.determineMethod(ctx);
    MethodDescriptor<?, ?> method = methodName != null ? methodsByName.get(methodName) : null;
    if (method == null) {
        // Unknown method, let the delegate return a usual error.
        return delegate().serve(ctx, req);
    }

    if (method.getType() != MethodType.UNARY) {
        return HttpResponse.of(HttpStatus.BAD_REQUEST,
                               MediaType.PLAIN_TEXT_UTF_8,
                               "Only unary methods can be used with non-framed requests.");
    }

    HttpHeaders grpcHeaders = HttpHeaders.copyOf(clientHeaders);

    final MediaType framedContentType;
    if (contentType.is(MediaType.PROTOBUF)) {
        framedContentType = GrpcSerializationFormats.PROTO.mediaType();
    } else if (contentType.is(MediaType.JSON_UTF_8)) {
        framedContentType = GrpcSerializationFormats.JSON.mediaType();
    } else {
        return HttpResponse.of(HttpStatus.UNSUPPORTED_MEDIA_TYPE,
                               MediaType.PLAIN_TEXT_UTF_8,
                               "Unsupported media type. Only application/protobuf is supported.");
    }
    grpcHeaders.contentType(framedContentType);

    if (grpcHeaders.get(GrpcHeaderNames.GRPC_ENCODING) != null) {
        return HttpResponse.of(HttpStatus.UNSUPPORTED_MEDIA_TYPE,
                               MediaType.PLAIN_TEXT_UTF_8,
                               "gRPC encoding is not supported for non-framed requests.");
    }

    // All clients support no encoding, and we don't support gRPC encoding for non-framed requests, so just
    // clear the header if it's present.
    grpcHeaders.remove(GrpcHeaderNames.GRPC_ACCEPT_ENCODING);

    final CompletableFuture<HttpResponse> responseFuture = new CompletableFuture<>();
    final HttpResponse res = HttpResponse.from(responseFuture);
    req.aggregate().whenCompleteAsync(
            (clientRequest, t) -> {
                if (t != null) {
                    responseFuture.completeExceptionally(t);
                } else {
                    frameAndServe(ctx, grpcHeaders, clientRequest, responseFuture);
                }
            },
            ctx.eventLoop());
    return res;
}

@Test
public void serverStreamingHeadersShouldNotBeFlushed() throws Exception {
  method = method.toBuilder().setType(MethodType.SERVER_STREAMING).build();
  shouldHeadersBeFlushed(false);
  shutdownAndVerify();
}

@Test
public void clientStreamingHeadersShouldBeFlushed() throws Exception {
  method = method.toBuilder().setType(MethodType.CLIENT_STREAMING).build();
  shouldHeadersBeFlushed(true);
  shutdownAndVerify();
}

@Test
public void duplexStreamingHeadersShouldNotBeFlushed() throws Exception {
  method = method.toBuilder().setType(MethodType.BIDI_STREAMING).build();
  shouldHeadersBeFlushed(true);
  shutdownAndVerify();
}

@Test
public void getType() {
  assertEquals(MethodType.UNARY, stream.getType());
}