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

private void processFileDescriptor(FileDescriptor fd,
        Map<String, FileDescriptor> descriptorsByName,
        Map<String, FileDescriptor> descriptorsBySymbol,
        Map<String, Map<Integer, FileDescriptor>> descriptorsByExtensionAndNumber) {
    String name = fd.getName();
    if (descriptorsByName.containsKey(name)) {
        throw new IllegalStateException("File name already used: " + name);
    }
    descriptorsByName.put(name, fd);
    for (Descriptors.ServiceDescriptor service : fd.getServices()) {
        processService(service, fd, descriptorsBySymbol);
    }
    for (Descriptors.Descriptor type : fd.getMessageTypes()) {
        processType(type, fd, descriptorsBySymbol, descriptorsByExtensionAndNumber);
    }
    for (Descriptors.FieldDescriptor extension : fd.getExtensions()) {
        processExtension(extension, fd, descriptorsByExtensionAndNumber);
    }
}
 

private void processService(Descriptors.ServiceDescriptor service, FileDescriptor fd,
        Map<String, FileDescriptor> descriptorsBySymbol) {
    String fullyQualifiedServiceName = service.getFullName();
    if (descriptorsBySymbol.containsKey(fullyQualifiedServiceName)) {
        throw new IllegalStateException("Service already defined: " + fullyQualifiedServiceName);
    }
    descriptorsBySymbol.put(fullyQualifiedServiceName, fd);
    for (Descriptors.MethodDescriptor method : service.getMethods()) {
        String fullyQualifiedMethodName = method.getFullName();
        if (descriptorsBySymbol.containsKey(fullyQualifiedMethodName)) {
            throw new IllegalStateException(
                    "Method already defined: " + fullyQualifiedMethodName + " in " + fullyQualifiedServiceName);
        }
        descriptorsBySymbol.put(fullyQualifiedMethodName, fd);
    }
}
 

@Bean
@Lazy
InfoContributor grpcInfoContributor(final GrpcServerProperties properties,
        final Collection<BindableService> grpcServices, final HealthStatusManager healthStatusManager) {
    final Map<String, Object> details = new LinkedHashMap<>();
    details.put("port", properties.getPort());

    if (properties.isReflectionServiceEnabled()) {
        // Only expose services via web-info if we do the same via grpc.
        final Map<String, List<String>> services = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
        details.put("services", services);
        final List<BindableService> mutableGrpcServiceList = new ArrayList<>(grpcServices);
        mutableGrpcServiceList.add(ProtoReflectionService.newInstance());
        if (properties.isHealthServiceEnabled()) {
            mutableGrpcServiceList.add(healthStatusManager.getHealthService());
        }
        for (final BindableService grpcService : mutableGrpcServiceList) {
            final ServiceDescriptor serviceDescriptor = grpcService.bindService().getServiceDescriptor();

            final List<String> methods = collectMethodNamesForService(serviceDescriptor);
            services.put(serviceDescriptor.getName(), methods);
        }
    }

    return new SimpleInfoContributor("grpc.server", details);
}
 

MethodHandlersBuilder(Object reactorService,
                      ServiceDescriptor serviceDefinition,
                      Class<CONTEXT> contextType,
                      Supplier<CONTEXT> contextResolver,
                      Class<REACT_SERVICE> reactorDetailedFallbackClass) {
    // CGLIB proxies do not retain generic type info. For these proxies we rely on a detailed fallback class definition to derive generic type info.
    Stream<Method> methodStream = AopUtils.isCglibProxy(reactorService) ? Stream.of(reactorDetailedFallbackClass.getMethods()) : Stream.of(reactorService.getClass().getMethods());
    this.reactorMethodMap = methodStream
            .filter(m -> !ReflectionExt.isObjectMethod(m))
            .collect(Collectors.toMap(Method::getName, Function.identity()));
    this.contextType = contextType;

    List<UnaryMethodHandler> unaryMethodHandlers = new ArrayList<>();
    List<ServerStreamingMethodHandler> serverStreamingMethodHandlers = new ArrayList<>();
    serviceDefinition.getMethods().forEach(methodDescriptor -> {
        GrpcToReactorMethodBinding binding = findReactorMethod(methodDescriptor);
        if (binding.isMono()) {
            unaryMethodHandlers.add(new UnaryMethodHandler<>(binding, contextResolver, reactorService));
        } else {
            serverStreamingMethodHandlers.add(new ServerStreamingMethodHandler<>(binding, contextResolver, reactorService));
        }
    });

    this.unaryMethodHandlers = unaryMethodHandlers;
    this.serverStreamingMethodHandlers = serverStreamingMethodHandlers;
}
 

@Bean
@Lazy
InfoContributor grpcInfoContributor(final GrpcServerProperties properties,
        final Collection<BindableService> grpcServices, final HealthStatusManager healthStatusManager) {
    final Map<String, Object> details = new LinkedHashMap<>();
    details.put("port", properties.getPort());

    if (properties.isReflectionServiceEnabled()) {
        // Only expose services via web-info if we do the same via grpc.
        final Map<String, List<String>> services = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
        details.put("services", services);
        final List<BindableService> mutableGrpcServiceList = new ArrayList<>(grpcServices);
        mutableGrpcServiceList.add(ProtoReflectionService.newInstance());
        if (properties.isHealthServiceEnabled()) {
            mutableGrpcServiceList.add(healthStatusManager.getHealthService());
        }
        for (final BindableService grpcService : mutableGrpcServiceList) {
            final ServiceDescriptor serviceDescriptor = grpcService.bindService().getServiceDescriptor();

            final List<String> methods = collectMethodNamesForService(serviceDescriptor);
            services.put(serviceDescriptor.getName(), methods);
        }
    }

    return new SimpleInfoContributor("grpc.server", details);
}
 

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

@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 ServiceDescriptor getServiceDescriptor(ServerServiceDefinition template, ProviderConfig providerConfig,
                                               List<MethodDescriptor<Request, Response>> methodDescriptors) {
    String serviceName = providerConfig.getInterfaceId();
    ServiceDescriptor.Builder builder = ServiceDescriptor.newBuilder(serviceName)
        .setSchemaDescriptor(template.getServiceDescriptor().getSchemaDescriptor());
    for (MethodDescriptor<Request, Response> methodDescriptor : methodDescriptors) {
        builder.addMethod(methodDescriptor);
    }
    return builder.build();

}
 

/**
 * Gets all method names from the given service descriptor.
 *
 * @param serviceDescriptor The service descriptor to get the names from.
 * @return The newly created and sorted list of the method names.
 */
protected List<String> collectMethodNamesForService(final ServiceDescriptor serviceDescriptor) {
    final List<String> methods = new ArrayList<>();
    for (final MethodDescriptor<?, ?> grpcMethod : serviceDescriptor.getMethods()) {
        methods.add(extractMethodName(grpcMethod));
    }
    methods.sort(String.CASE_INSENSITIVE_ORDER);
    return methods;
}
 

public GrpcToReactorClientFactory getGrpcToReactorClientFactory() {
    return new GrpcToReactorClientFactory() {
        @Override
        public <GRPC_STUB extends AbstractStub<GRPC_STUB>, REACT_API> REACT_API apply(GRPC_STUB stub, Class<REACT_API> apiType, ServiceDescriptor serviceDescriptor) {
            return ReactorToGrpcClientBuilder
                    .<REACT_API, GRPC_STUB, CallMetadata>newBuilder(
                            apiType, stub, serviceDescriptor, CallMetadata.class
                    )
                    .withGrpcStubDecorator(CommonCallMetadataUtils.newGrpcStubDecorator(AnonymousCallMetadataResolver.getInstance()))
                    .withTimeout(Duration.ofMillis(GrpcRequestConfiguration.DEFAULT_REQUEST_TIMEOUT_MS))
                    .withStreamingTimeout(Duration.ofMillis(GrpcRequestConfiguration.DEFAULT_STREAMING_TIMEOUT_MS))
                    .build();
        }
    };
}
 

/**
 * Override to add server side interceptors.
 */
protected List<ServerInterceptor> createInterceptors(ServiceDescriptor serviceDescriptor) {
    List<ServerInterceptor> interceptors = new ArrayList<ServerInterceptor>();
    interceptors.add(admissionControllerServerInterceptor);
    interceptors.add(new ErrorCatchingServerInterceptor());
    interceptors.add(leaderServerInterceptor);
    interceptors.add(new V3HeaderInterceptor());
    return GrpcFitInterceptor.appendIfFitEnabled(interceptors, titusRuntime);
}
 

/**
 * Override to add server side interceptors.
 */
protected List<ServerInterceptor> createInterceptors(ServiceDescriptor serviceDescriptor) {
    return Arrays.asList(
            admissionController,
            new ErrorCatchingServerInterceptor(),
            new V3HeaderInterceptor()
    );
}
 

@Override
public <GRPC_STUB extends AbstractStub<GRPC_STUB>, REACT_API> REACT_API apply(GRPC_STUB stub, Class<REACT_API> apiType, ServiceDescriptor serviceDescriptor) {
    return ReactorToGrpcClientBuilder
            .newBuilder(
                    apiType, stub, serviceDescriptor, contextType
            )
            .withGrpcStubDecorator((BiFunction<GRPC_STUB, Optional<CONTEXT>, GRPC_STUB>) grpcStubDecorator)
            .withTimeout(Duration.ofMillis(configuration.getRequestTimeoutMs()))
            .withStreamingTimeout(Duration.ofMillis(configuration.getStreamingTimeoutMs()))
            .build();
}
 

public GrpcLeaderServerInterceptor(LeaderActivationStatus leaderActivationStatus,
                                   List<ServiceDescriptor> notProtectedServices) {
    this.leaderActivationStatus = leaderActivationStatus;
    this.notProtectedMethods = notProtectedServices.stream()
            .flatMap(s -> s.getMethods().stream())
            .collect(Collectors.toCollection(() -> new TreeSet<>(new MethodDescriptorComparator())));
}
 

private MethodDescriptor getMethod(ServiceDescriptor serviceDescriptor, String methodName) {
    MethodDescriptor<?, ?> methodDescriptor = serviceDescriptor.getMethods().stream()
            .filter(m -> m.getFullMethodName().contains(methodName))
            .findFirst()
            .orElseThrow(() -> new IllegalArgumentException("Method not found: " + methodName));

    // Rebuild to get a different version
    return methodDescriptor.toBuilder().build();
}
 

/**
 * Override to add server side interceptors.
 */
protected List<ServerInterceptor> createInterceptors(ServiceDescriptor serviceDescriptor) {
    return GrpcFitInterceptor.appendIfFitEnabled(
            asList(new ErrorCatchingServerInterceptor(), new V3HeaderInterceptor()),
            titusRuntime
    );
}
 

private ReactorToGrpcClientBuilder(Class<REACT_API> reactApi,
                                   GRPC_STUB grpcStub,
                                   ServiceDescriptor grpcServiceDescriptor,
                                   Class<CONTEXT> contextType) {
    this.reactApi = reactApi;
    this.grpcStub = grpcStub;
    this.grpcServiceDescriptor = grpcServiceDescriptor;
    this.contextType = contextType;
    this.nonGrpcParameters = Collections.singleton(this.contextType);
}
 

public static <REACT_API, GRPC_STUB extends AbstractStub<GRPC_STUB>, CONTEXT> ReactorToGrpcClientBuilder<REACT_API, GRPC_STUB, CONTEXT> newBuilder(
        Class<REACT_API> reactApi,
        GRPC_STUB grpcStub,
        ServiceDescriptor grpcServiceDescriptor,
        Class<CONTEXT> contextType) {
    Preconditions.checkArgument(reactApi.isInterface(), "Interface type required");
    return new ReactorToGrpcClientBuilder<>(reactApi, grpcStub, grpcServiceDescriptor, contextType);
}
 

public static <REACT_API, GRPC_STUB extends AbstractStub<GRPC_STUB>, CONTEXT> ReactorToGrpcClientBuilder<REACT_API, GRPC_STUB, CONTEXT> newBuilderWithDefaults(
        Class<REACT_API> reactApi,
        GRPC_STUB grpcStub,
        ServiceDescriptor grpcServiceDescriptor,
        Class<CONTEXT> contextType) {
    Preconditions.checkArgument(reactApi.isInterface(), "Interface type required");
    return new ReactorToGrpcClientBuilder<>(reactApi, grpcStub, grpcServiceDescriptor, contextType)
            .withTimeout(Duration.ofMillis(DEFAULT_REQUEST_TIMEOUT_MS))
            .withStreamingTimeout(Duration.ofMillis(DEFAULT_STREAMING_TIMEOUT_MS))
            .withGrpcStubDecorator(EMPTY_STUB_DECORATOR);
}
 

/**
 * If grpcArgPos is less then zero, it means no GRPC argument is provided, and instead {@link Empty} value should be used.
 * If contextPos is less then zero, it means the context value should be resolved as it is not passed directly by
 * the client.
 */
FluxMethodBridge(Method reactMethod,
                 ServiceDescriptor grpcServiceDescriptor,
                 Method grpcMethod,
                 int grpcArgPos,
                 int contextPos,
                 BiFunction<GRPC_STUB, Optional<CONTEXT>, GRPC_STUB> grpcStubDecorator,
                 GRPC_STUB grpcStub,
                 Duration timeout,
                 Duration streamingTimeout) {
    this.grpcArgPos = grpcArgPos;
    this.contextPos = contextPos;
    this.grpcStubDecorator = grpcStubDecorator;
    this.grpcStub = grpcStub;

    this.streamingResponse = grpcServiceDescriptor.getMethods().stream()
            .filter(m -> toMethodNameFromFullName(m.getFullMethodName()).equals(reactMethod.getName()))
            .findFirst()
            .map(m -> m.getType() == MethodDescriptor.MethodType.SERVER_STREAMING)
            .orElse(false);

    Preconditions.checkArgument(
            !GrpcToReactUtil.isEmptyToVoidResult(reactMethod, grpcMethod),
            "Empty GRPC reply to Flux<Mono> mapping not supported (use Mono<Void> in API definition instead)"
    );
    this.grpcMethod = grpcMethod;
    this.timeout = streamingResponse ? streamingTimeout : timeout;
    this.reactorTimeout = Duration.ofMillis((long) (timeout.toMillis() * GrpcToReactUtil.RX_CLIENT_TIMEOUT_FACTOR));
}
 

@Override
public <REACT_SERVICE> ServerServiceDefinition apply(ServiceDescriptor serviceDescriptor, REACT_SERVICE reactService, Class<REACT_SERVICE> reactorDetailedFallbackClass) {
    return GrpcToReactorServerBuilder.<REACT_SERVICE, CONTEXT>newBuilder(serviceDescriptor, reactService)
            .withContext(contextType, contextResolver)
            .withReactorFallbackClass(reactorDetailedFallbackClass)
            .build();
}
 

/**
 * Gets all method names from the given service descriptor.
 *
 * @param serviceDescriptor The service descriptor to get the names from.
 * @return The newly created and sorted list of the method names.
 */
protected List<String> collectMethodNamesForService(final ServiceDescriptor serviceDescriptor) {
    final List<String> methods = new ArrayList<>();
    for (final MethodDescriptor<?, ?> grpcMethod : serviceDescriptor.getMethods()) {
        methods.add(extractMethodName(grpcMethod));
    }
    methods.sort(String.CASE_INSENSITIVE_ORDER);
    return methods;
}
 

@Override
public Server createServer() {
	ServerBuilder builder = ServerBuilder.forPort(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();
}
 

/**
 * Returns a newly-created {@link GrpcJsonMarshaller} with the specified {@link ServiceDescriptor}.
 */
public GrpcJsonMarshaller build(ServiceDescriptor serviceDescriptor) {
    requireNonNull(serviceDescriptor, "serviceDescriptor");
    return new DefaultJsonMarshaller(
            GrpcJsonUtil.jsonMarshaller(ImmutableList.copyOf(serviceDescriptor.getMethods()),
                                        jsonMarshallerCustomizer));
}
 

@Test
void json_preservingFieldNames() throws Exception {
    final AtomicReference<HttpHeaders> requestHeaders = new AtomicReference<>();
    final AtomicReference<byte[]> payload = new AtomicReference<>();
    final Function<ServiceDescriptor, GrpcJsonMarshaller> marshallerFactory = serviceDescriptor -> {
        return GrpcJsonMarshaller.builder()
                                 .jsonMarshallerCustomizer(marshaller -> {
                                     marshaller.preservingProtoFieldNames(true);
                                 })
                                 .build(serviceDescriptor);
    };
    final UnitTestServiceBlockingStub jsonStub =
            Clients.builder(server.httpUri(GrpcSerializationFormats.JSON) + "/json-preserving/")
                   .option(GrpcClientOptions.GRPC_JSON_MARSHALLER_FACTORY.newValue(marshallerFactory))
                   .decorator(client -> new SimpleDecoratingHttpClient(client) {
                       @Override
                       public HttpResponse execute(ClientRequestContext ctx, HttpRequest req)
                               throws Exception {
                           requestHeaders.set(req.headers());
                           return new FilteredHttpResponse(unwrap().execute(ctx, req)) {
                               @Override
                               protected HttpObject filter(HttpObject obj) {
                                   if (obj instanceof HttpData) {
                                       payload.set(((HttpData) obj).array());
                                   }
                                   return obj;
                               }
                           };
                       }
                   })
                   .build(UnitTestServiceBlockingStub.class);
    final SimpleResponse response = jsonStub.staticUnaryCall(REQUEST_MESSAGE);
    assertThat(response).isEqualTo(RESPONSE_MESSAGE);
    assertThat(requestHeaders.get().get(HttpHeaderNames.CONTENT_TYPE)).isEqualTo(
            "application/grpc+json");

    final byte[] deframed = Arrays.copyOfRange(payload.get(), 5, payload.get().length);
    assertThat(new String(deframed, StandardCharsets.UTF_8)).contains("oauth_scope");
}