Java 类io.grpc.ServiceDescriptor 实例源码

@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 Server createServer() {
    NettyServerBuilder builder = NettyServerBuilder.forAddress(
            new InetSocketAddress(InetAddresses.forString(getAddress()), getPort()));
    Collection<GrpcServiceDefinition> definitions = discoverer.findGrpcServices();
    for (GrpcServiceDefinition definition : definitions) {
        ServiceDescriptor descriptor = definition.getService().bindService().getServiceDescriptor();
        logger.info("Registered gRPC service: " + descriptor.getName()
                + ", bean: " + definition.getBeanName() + ", class: "
                + definition.getService().getClass().getName());
        builder.addService(definition.getService());
    }

    return builder.build();
}

LoadServer(Control.ServerConfig config) throws Exception {
  log.log(Level.INFO, "Server Config \n" + config.toString());
  port = config.getPort() ==  0 ? Utils.pickUnusedPort() : config.getPort();
  ServerBuilder<?> serverBuilder = ServerBuilder.forPort(port);
  int asyncThreads = config.getAsyncServerThreads() == 0
      ? Runtime.getRuntime().availableProcessors()
      : config.getAsyncServerThreads();
  // The concepts of sync & async server are quite different in the C impl and the names
  // chosen for the enum are based on that implementation. We use 'sync' to mean
  // the direct executor case in Java even though the service implementations are always
  // fully async.
  switch (config.getServerType()) {
    case ASYNC_SERVER: {
      serverBuilder.executor(getExecutor(asyncThreads));
      break;
    }
    case SYNC_SERVER: {
      serverBuilder.directExecutor();
      break;
    }
    case ASYNC_GENERIC_SERVER: {
      serverBuilder.executor(getExecutor(asyncThreads));
      // Create buffers for the generic service
      PooledByteBufAllocator alloc = PooledByteBufAllocator.DEFAULT;
      genericResponse = alloc.buffer(config.getPayloadConfig().getBytebufParams().getRespSize());
      if (genericResponse.capacity() > 0) {
        genericResponse.writerIndex(genericResponse.capacity() - 1);
      }
      break;
    }
    default: {
      throw new IllegalArgumentException();
    }
  }
  if (config.hasSecurityParams()) {
    File cert = TestUtils.loadCert("server1.pem");
    File key = TestUtils.loadCert("server1.key");
    serverBuilder.useTransportSecurity(cert, key);
  }
  benchmarkService = new AsyncServer.BenchmarkServiceImpl();
  if (config.getServerType() == Control.ServerType.ASYNC_GENERIC_SERVER) {
    serverBuilder.addService(
        ServerServiceDefinition
            .builder(new ServiceDescriptor(BenchmarkServiceGrpc.SERVICE_NAME,
                GENERIC_STREAMING_PING_PONG_METHOD))
            .addMethod(GENERIC_STREAMING_PING_PONG_METHOD, new GenericServiceCallHandler())
            .build());
  } else {
    serverBuilder.addService(benchmarkService);
  }
  server = serverBuilder.build();

  List<OperatingSystemMXBean> beans =
      ManagementFactory.getPlatformMXBeans(OperatingSystemMXBean.class);
  if (!beans.isEmpty()) {
    osBean = beans.get(0);
  } else {
    osBean = null;
  }
}

@Test
public void exceptionInStartCallPropagatesToStream() throws Exception {
  createAndStartServer();
  final Status status = Status.ABORTED.withDescription("Oh, no!");
  mutableFallbackRegistry.addService(ServerServiceDefinition.builder(
      new ServiceDescriptor("Waiter", METHOD))
      .addMethod(METHOD,
          new ServerCallHandler<String, Integer>() {
            @Override
            public ServerCall.Listener<String> startCall(
                ServerCall<String, Integer> call,
                Metadata headers) {
              throw status.asRuntimeException();
            }
          }).build());
  ServerTransportListener transportListener
      = transportServer.registerNewServerTransport(new SimpleServerTransport());
  transportListener.transportReady(Attributes.EMPTY);

  Metadata requestHeaders = new Metadata();
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newServerContext(streamTracerFactories, "Waiter/serve", requestHeaders);
  when(stream.statsTraceContext()).thenReturn(statsTraceCtx);

  transportListener.streamCreated(stream, "Waiter/serve", requestHeaders);
  verify(stream).setListener(streamListenerCaptor.capture());
  ServerStreamListener streamListener = streamListenerCaptor.getValue();
  assertNotNull(streamListener);
  verify(stream, atLeast(1)).statsTraceContext();
  verifyNoMoreInteractions(stream);
  verify(fallbackRegistry, never()).lookupMethod(any(String.class), any(String.class));

  assertEquals(1, executor.runDueTasks());
  verify(fallbackRegistry).lookupMethod("Waiter/serve", AUTHORITY);
  verify(stream).close(same(status), notNull(Metadata.class));
  verify(stream, atLeast(1)).statsTraceContext();
}

@Test
public void handlerRegistryPriorities() throws Exception {
  fallbackRegistry = mock(HandlerRegistry.class);
  builder.addService(
      ServerServiceDefinition.builder(new ServiceDescriptor("Waiter", METHOD))
          .addMethod(METHOD, callHandler).build());
  transportServer = new SimpleServer();
  createAndStartServer();

  ServerTransportListener transportListener
      = transportServer.registerNewServerTransport(new SimpleServerTransport());
  transportListener.transportReady(Attributes.EMPTY);
  Metadata requestHeaders = new Metadata();
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newServerContext(streamTracerFactories, "Waiter/serve", requestHeaders);
  when(stream.statsTraceContext()).thenReturn(statsTraceCtx);

  // This call will be handled by callHandler from the internal registry
  transportListener.streamCreated(stream, "Waiter/serve", requestHeaders);
  assertEquals(1, executor.runDueTasks());
  verify(callHandler).startCall(Matchers.<ServerCall<String, Integer>>anyObject(),
      Matchers.<Metadata>anyObject());
  // This call will be handled by the fallbackRegistry because it's not registred in the internal
  // registry.
  transportListener.streamCreated(stream, "Service1/Method2", requestHeaders);
  assertEquals(1, executor.runDueTasks());
  verify(fallbackRegistry).lookupMethod("Service1/Method2", AUTHORITY);

  verifyNoMoreInteractions(callHandler);
  verifyNoMoreInteractions(fallbackRegistry);
}

@Test
public void addReturnsPrevious() {
  assertNull(registry.addService(basicServiceDefinition));
  assertSame(basicServiceDefinition,
      registry.addService(ServerServiceDefinition.builder(
          new ServiceDescriptor("basic")).build()));
}

private ServerStreamListener testClientClose_setup(
    final AtomicReference<ServerCall<String, Integer>> callReference,
    final AtomicReference<Context> context,
    final AtomicBoolean contextCancelled) throws Exception {
  createAndStartServer();
  callListener = new ServerCall.Listener<String>() {
    @Override
    public void onReady() {
      context.set(Context.current());
      Context.current().addListener(new Context.CancellationListener() {
        @Override
        public void cancelled(Context context) {
          contextCancelled.set(true);
        }
      }, MoreExecutors.directExecutor());
    }
  };

  mutableFallbackRegistry.addService(ServerServiceDefinition.builder(
      new ServiceDescriptor("Waiter", METHOD))
      .addMethod(METHOD,
          new ServerCallHandler<String, Integer>() {
            @Override
            public ServerCall.Listener<String> startCall(
                ServerCall<String, Integer> call,
                Metadata headers) {
              callReference.set(call);
              return callListener;
            }
          }).build());
  ServerTransportListener transportListener
      = transportServer.registerNewServerTransport(new SimpleServerTransport());
  transportListener.transportReady(Attributes.EMPTY);
  Metadata requestHeaders = new Metadata();
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newServerContext(streamTracerFactories, "Waitier/serve", requestHeaders);
  when(stream.statsTraceContext()).thenReturn(statsTraceCtx);
  transportListener.streamCreated(stream, "Waiter/serve", requestHeaders);
  verify(stream).setListener(streamListenerCaptor.capture());
  ServerStreamListener streamListener = streamListenerCaptor.getValue();
  assertNotNull(streamListener);

  streamListener.onReady();
  assertEquals(1, executor.runDueTasks());
  return streamListener;
}

@Test
public void removeMissingNameConflictFails() {
  assertNull(registry.addService(basicServiceDefinition));
  assertFalse(registry.removeService(ServerServiceDefinition.builder(
      new ServiceDescriptor("basic")).build()));
}

@Test
public void inprocessTransportManualFlow() throws Exception {
  final Semaphore semaphore = new Semaphore(1);
  ServerServiceDefinition service = ServerServiceDefinition.builder(
      new ServiceDescriptor("some", STREAMING_METHOD))
      .addMethod(STREAMING_METHOD, ServerCalls.asyncBidiStreamingCall(
          new ServerCalls.BidiStreamingMethod<Integer, Integer>() {
            int iteration;

            @Override
            public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) {
              final ServerCallStreamObserver<Integer> serverCallObserver =
                  (ServerCallStreamObserver<Integer>) responseObserver;
              serverCallObserver.setOnReadyHandler(new Runnable() {
                @Override
                public void run() {
                  while (serverCallObserver.isReady()) {
                    serverCallObserver.onNext(iteration);
                  }
                  iteration++;
                  semaphore.release();
                }
              });
              return new ServerCalls.NoopStreamObserver<Integer>() {
                @Override
                public void onCompleted() {
                  serverCallObserver.onCompleted();
                }
              };
            }
          }))
      .build();
  long tag = System.nanoTime();
  InProcessServerBuilder.forName("go-with-the-flow" + tag).addService(service).build().start();
  ManagedChannel channel = InProcessChannelBuilder.forName("go-with-the-flow" + tag).build();
  final ClientCall<Integer, Integer> clientCall = channel.newCall(STREAMING_METHOD,
      CallOptions.DEFAULT);
  final CountDownLatch latch = new CountDownLatch(1);
  final int[] receivedMessages = new int[6];
  clientCall.start(new ClientCall.Listener<Integer>() {
    int index;

    @Override
    public void onMessage(Integer message) {
      receivedMessages[index++] = message;
    }

    @Override
    public void onClose(Status status, Metadata trailers) {
      latch.countDown();
    }
  }, new Metadata());
  semaphore.acquire();
  clientCall.request(1);
  semaphore.acquire();
  clientCall.request(2);
  semaphore.acquire();
  clientCall.request(3);
  clientCall.halfClose();
  latch.await(5, TimeUnit.SECONDS);
  // Very that number of messages produced in each onReady handler call matches the number
  // requested by the client.
  assertArrayEquals(new int[]{0, 1, 1, 2, 2, 2}, receivedMessages);
}