Java 类io.grpc.CompressorRegistry 实例源码

GrpcService(HandlerRegistry registry,
            Set<PathMapping> pathMappings,
            DecompressorRegistry decompressorRegistry,
            CompressorRegistry compressorRegistry,
            Set<SerializationFormat> supportedSerializationFormats,
            int maxOutboundMessageSizeBytes,
            int maxInboundMessageSizeBytes) {
    this.registry = requireNonNull(registry, "registry");
    this.pathMappings = requireNonNull(pathMappings, "pathMappings");
    this.decompressorRegistry = requireNonNull(decompressorRegistry, "decompressorRegistry");
    this.compressorRegistry = requireNonNull(compressorRegistry, "compressorRegistry");
    this.supportedSerializationFormats = supportedSerializationFormats;
    jsonMarshaller = jsonMarshaller(registry);
    this.maxOutboundMessageSizeBytes = maxOutboundMessageSizeBytes;
    this.maxInboundMessageSizeBytes = maxInboundMessageSizeBytes;
}

/**
 * Constructs a new {@link GrpcService} that can be bound to
 * {@link com.linecorp.armeria.server.ServerBuilder}. It is recommended to bind the service to a server
 * using {@link com.linecorp.armeria.server.ServerBuilder#service(ServiceWithPathMappings)} to mount all
 * service paths without interfering with other services.
 */
public ServiceWithPathMappings<HttpRequest, HttpResponse> build() {
    HandlerRegistry handlerRegistry = registryBuilder.build();
    GrpcService grpcService = new GrpcService(
            handlerRegistry,
            handlerRegistry
                  .methods()
                  .keySet()
                  .stream()
                  .map(path -> PathMapping.ofExact("/" + path))
                  .collect(ImmutableSet.toImmutableSet()),
            firstNonNull(decompressorRegistry, DecompressorRegistry.getDefaultInstance()),
            firstNonNull(compressorRegistry, CompressorRegistry.getDefaultInstance()),
            supportedSerializationFormats, maxOutboundMessageSizeBytes,
            maxInboundMessageSizeBytes);
    return enableUnframedRequests ? grpcService.decorate(UnframedGrpcService::new) : grpcService;
}

@Before
public void setUp() {
    when(ctx.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
    call = new ArmeriaServerCall<>(
            HttpHeaders.of(),
            TestServiceGrpc.METHOD_UNARY_CALL,
            CompressorRegistry.getDefaultInstance(),
            DecompressorRegistry.getDefaultInstance(),
            res,
            MAX_MESSAGE_BYTES,
            MAX_MESSAGE_BYTES,
            ctx,
            GrpcSerializationFormats.PROTO,
            MessageMarshaller.builder().build());
    call.setListener(listener);
    call.messageReader().onSubscribe(subscription);
    when(ctx.logBuilder()).thenReturn(new DefaultRequestLog(ctx));
    when(ctx.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
}

private void sendMessage_serverSendsOne_closeOnSecondCall(
    MethodDescriptor<Long, Long> method) {
  ServerCallImpl<Long, Long> serverCall = new ServerCallImpl<Long, Long>(
      stream,
      method,
      requestHeaders,
      context,
      DecompressorRegistry.getDefaultInstance(),
      CompressorRegistry.getDefaultInstance());
  serverCall.sendHeaders(new Metadata());
  serverCall.sendMessage(1L);
  verify(stream, times(1)).writeMessage(any(InputStream.class));
  verify(stream, never()).close(any(Status.class), any(Metadata.class));

  // trying to send a second message causes gRPC to close the underlying stream
  serverCall.sendMessage(1L);
  verify(stream, times(1)).writeMessage(any(InputStream.class));
  ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
  verify(stream, times(1)).close(statusCaptor.capture(), metadataCaptor.capture());
  assertEquals(Status.Code.INTERNAL, statusCaptor.getValue().getCode());
  assertEquals(ServerCallImpl.TOO_MANY_RESPONSES, statusCaptor.getValue().getDescription());
  assertTrue(metadataCaptor.getValue().keys().isEmpty());
}

private void sendMessage_serverSendsOne_closeOnSecondCall_appRunToCompletion(
    MethodDescriptor<Long, Long> method) {
  ServerCallImpl<Long, Long> serverCall = new ServerCallImpl<Long, Long>(
      stream,
      method,
      requestHeaders,
      context,
      DecompressorRegistry.getDefaultInstance(),
      CompressorRegistry.getDefaultInstance());
  serverCall.sendHeaders(new Metadata());
  serverCall.sendMessage(1L);
  serverCall.sendMessage(1L);
  verify(stream, times(1)).writeMessage(any(InputStream.class));
  verify(stream, times(1)).close(any(Status.class), any(Metadata.class));

  // App runs to completion but everything is ignored
  serverCall.sendMessage(1L);
  serverCall.close(Status.OK, new Metadata());
  try {
    serverCall.close(Status.OK, new Metadata());
    fail("calling a second time should still cause an error");
  } catch (IllegalStateException expected) {
    // noop
  }
}

private void serverSendsOne_okFailsOnMissingResponse(
    MethodDescriptor<Long, Long> method) {
  ServerCallImpl<Long, Long> serverCall = new ServerCallImpl<Long, Long>(
      stream,
      method,
      requestHeaders,
      context,
      DecompressorRegistry.getDefaultInstance(),
      CompressorRegistry.getDefaultInstance());
  serverCall.close(Status.OK, new Metadata());
  ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
  verify(stream, times(1)).close(statusCaptor.capture(), metadataCaptor.capture());
  assertEquals(Status.Code.INTERNAL, statusCaptor.getValue().getCode());
  assertEquals(ServerCallImpl.MISSING_RESPONSE, statusCaptor.getValue().getDescription());
  assertTrue(metadataCaptor.getValue().keys().isEmpty());
}

ArmeriaClientCall(
        ClientRequestContext ctx,
        Client<HttpRequest, HttpResponse> httpClient,
        HttpRequestWriter req,
        MethodDescriptor<I, O> method,
        int maxOutboundMessageSizeBytes,
        int maxInboundMessageSizeBytes,
        CallOptions callOptions,
        CompressorRegistry compressorRegistry,
        DecompressorRegistry decompressorRegistry,
        SerializationFormat serializationFormat,
        @Nullable MessageMarshaller jsonMarshaller) {
    this.ctx = ctx;
    this.httpClient = httpClient;
    this.req = req;
    this.callOptions = callOptions;
    this.compressorRegistry = compressorRegistry;
    this.decompressorRegistry = decompressorRegistry;
    this.messageFramer = new ArmeriaMessageFramer(ctx.alloc(), maxOutboundMessageSizeBytes);
    this.marshaller = new GrpcMessageMarshaller<>(
            ctx.alloc(), serializationFormat, method, jsonMarshaller);
    responseReader = new HttpStreamReader(
            decompressorRegistry,
            new ArmeriaMessageDeframer(this, maxInboundMessageSizeBytes, ctx.alloc()),
            this);
    executor = callOptions.getExecutor();
}

@Override
public <I, O> ClientCall<I, O> newCall(
        MethodDescriptor<I, O> method, CallOptions callOptions) {
    HttpRequestWriter req = HttpRequest.streaming(
            HttpHeaders
                    .of(HttpMethod.POST, uri().getPath() + method.getFullMethodName())
                    .contentType(serializationFormat.mediaType()));
    ClientRequestContext ctx = newContext(HttpMethod.POST, req);
    ctx.logBuilder().serializationFormat(serializationFormat);
    ctx.logBuilder().requestContent(GrpcLogUtil.rpcRequest(method), null);
    ctx.logBuilder().deferResponseContent();
    return new ArmeriaClientCall<>(
            ctx,
            httpClient,
            req,
            method,
            options().getOrElse(GrpcClientOptions.MAX_OUTBOUND_MESSAGE_SIZE_BYTES,
                                ArmeriaMessageFramer.NO_MAX_OUTBOUND_MESSAGE_SIZE),
            options().getOrElse(
                    GrpcClientOptions.MAX_INBOUND_MESSAGE_SIZE_BYTES,
                    options().getOrElse(
                            ClientOption.DEFAULT_MAX_RESPONSE_LENGTH,
                            (long) DEFAULT_MAX_INBOUND_MESSAGE_SIZE).intValue()),
            callOptions,
            CompressorRegistry.getDefaultInstance(),
            DecompressorRegistry.getDefaultInstance(),
            serializationFormat,
            jsonMarshaller);
}

ArmeriaServerCall(HttpHeaders clientHeaders,
                  MethodDescriptor<I, O> method,
                  CompressorRegistry compressorRegistry,
                  DecompressorRegistry decompressorRegistry,
                  HttpResponseWriter res,
                  int maxInboundMessageSizeBytes,
                  int maxOutboundMessageSizeBytes,
                  ServiceRequestContext ctx,
                  SerializationFormat serializationFormat,
                  MessageMarshaller jsonMarshaller) {
    requireNonNull(clientHeaders, "clientHeaders");
    this.method = requireNonNull(method, "method");
    this.ctx = requireNonNull(ctx, "ctx");
    this.serializationFormat = requireNonNull(serializationFormat, "serializationFormat");
    this.messageReader = new HttpStreamReader(
            requireNonNull(decompressorRegistry, "decompressorRegistry"),
            new ArmeriaMessageDeframer(
                    this,
                    maxInboundMessageSizeBytes,
                    ctx.alloc())
                    .decompressor(clientDecompressor(clientHeaders, decompressorRegistry)),
            this);
    this.messageFramer = new ArmeriaMessageFramer(ctx.alloc(), maxOutboundMessageSizeBytes);
    this.res = requireNonNull(res, "res");
    this.compressorRegistry = requireNonNull(compressorRegistry, "compressorRegistry");
    this.clientAcceptEncoding =
            Strings.emptyToNull(clientHeaders.get(GrpcHeaderNames.GRPC_ACCEPT_ENCODING));
    this.decompressorRegistry = requireNonNull(decompressorRegistry, "decompressorRegistry");
    marshaller = new GrpcMessageMarshaller<>(ctx.alloc(), serializationFormat, method, jsonMarshaller);
}

@Override
public final T compressorRegistry(CompressorRegistry registry) {
  if (registry != null) {
    this.compressorRegistry = registry;
  } else {
    this.compressorRegistry = DEFAULT_COMPRESSOR_REGISTRY;
  }
  return thisT();
}

ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method,
    Metadata inboundHeaders, Context.CancellableContext context,
    DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry) {
  this.stream = stream;
  this.method = method;
  this.context = context;
  this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);
  this.decompressorRegistry = decompressorRegistry;
  this.compressorRegistry = compressorRegistry;
}

@Override
public final T compressorRegistry(CompressorRegistry registry) {
  if (registry != null) {
    compressorRegistry = registry;
  } else {
    compressorRegistry = DEFAULT_COMPRESSOR_REGISTRY;
  }
  return thisT();
}

@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  context = Context.ROOT.withCancellation();
  call = new ServerCallImpl<Long, Long>(stream, UNARY_METHOD, requestHeaders, context,
      DecompressorRegistry.getDefaultInstance(), CompressorRegistry.getDefaultInstance());
}

@Test
public void compressorRegistry_normal() {
  CompressorRegistry compressorRegistry = CompressorRegistry.newEmptyInstance();
  assertNotEquals(compressorRegistry, builder.compressorRegistry);
  assertEquals(builder, builder.compressorRegistry(compressorRegistry));
  assertEquals(compressorRegistry, builder.compressorRegistry);
}

@Test
public void compressorRegistry_null() {
  CompressorRegistry defaultValue = builder.compressorRegistry;
  builder.compressorRegistry(CompressorRegistry.newEmptyInstance());
  assertNotEquals(defaultValue, builder.compressorRegistry);
  assertEquals(builder, builder.compressorRegistry(null));
  assertEquals(defaultValue, builder.compressorRegistry);
}

@Override
public DropwizardServerBuilder compressorRegistry(@Nullable final CompressorRegistry registry) {
    origin.compressorRegistry(registry);
    return this;
}

/**
 * Sets the {@link CompressorRegistry} to use when compressing messages. If not set, will use the
 * default, which supports gzip only.
 */
public GrpcServiceBuilder compressorRegistry(CompressorRegistry registry) {
    compressorRegistry = requireNonNull(registry, "registry");
    return this;
}

ClientCallImpl<ReqT, RespT> setCompressorRegistry(CompressorRegistry compressorRegistry) {
  this.compressorRegistry = compressorRegistry;
  return this;
}