io.grpc.services.HealthStatusManager#io.grpc.util.TransmitStatusRuntimeExceptionInterceptor源码实例Demo

@Override
public void start() throws Exception {
  profileStore = new LocalProfileStore(kvStoreProvider.get());
  profileStore.start();

  metricsStore = new LocalMetricsStore();
  metricsStore.start();

  server = JobTelemetryRpcUtils.newInProcessServerBuilder(grpcFactory,
    selfEndpoint.get().getFabricPort())
    .maxInboundMetadataSize(81920) // GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE * 10
    .intercept(TransmitStatusRuntimeExceptionInterceptor.instance())
    .addService(new JobTelemetryServiceImpl(metricsStore, profileStore, true))
    .build();

  server.start();
  logger.info("LocalJobTelemetryServer is up");
}
 

@Override
public void start() throws Exception {
  final ConduitServiceRegistryImpl registry = (ConduitServiceRegistryImpl) registryProvider.get();

  for (BindableService service : registry.getServiceList()) {
    serverBuilder.addService(service);
  }

  for (CloseableBindableService closeableService : registry.getCloseableServiceList()) {
    serverBuilder.addService(closeableService);
    closeableServices.add(closeableService);
  }

  serverBuilder.maxInboundMetadataSize(Integer.MAX_VALUE).maxInboundMessageSize(Integer.MAX_VALUE)
    .intercept(TransmitStatusRuntimeExceptionInterceptor.instance());

  if (sslEngineFactory.isPresent()) {
    final SslContextBuilder contextBuilder = sslEngineFactory.get().newServerContextBuilder();
    // add gRPC overrides using #configure
    serverBuilder.sslContext(GrpcSslContexts.configure(contextBuilder).build());
  }
  server = serverBuilder.build();
  server.start();

  logger.info("ConduitServer is up. Listening on port '{}'", server.getPort());
}
 

public HttpProxy(
    ServerBuilder<?> serverBuilder, @Nullable Credentials creds, HttpProxyOptions options)
    throws URISyntaxException, SSLException {
  super("HttpProxy");
  this.options = options;
  SimpleBlobStore simpleBlobStore =
      HttpBlobStore.create(
          URI.create(options.httpCache),
          /* remoteMaxConnections=*/ 0,
          (int) SECONDS.toMillis(options.timeout),
          creds);
  server =
      serverBuilder
          .addService(new ActionCacheService(simpleBlobStore))
          .addService(
              new ContentAddressableStorageService(
                  simpleBlobStore, options.treeDefaultPageSize, options.treeMaxPageSize))
          .addService(new ByteStreamService(simpleBlobStore))
          .intercept(TransmitStatusRuntimeExceptionInterceptor.instance())
          .build();
}
 

public BuildFarmServer(
    String session, ServerBuilder<?> serverBuilder, BuildFarmServerConfig config)
    throws InterruptedException, ConfigurationException {
  super("BuildFarmServer");
  String defaultInstanceName = config.getDefaultInstanceName();
  instances =
      new BuildFarmInstances(session, config.getInstancesList(), defaultInstanceName, this::stop);

  healthStatusManager = new HealthStatusManager();
  actionCacheRequestCounter =
      new ActionCacheRequestCounter(ActionCacheService.logger, Duration.ofSeconds(10));

  ServerInterceptor headersInterceptor = new ServerHeadersInterceptor();

  server =
      serverBuilder
          .addService(healthStatusManager.getHealthService())
          .addService(new ActionCacheService(instances, actionCacheRequestCounter::increment))
          .addService(new CapabilitiesService(instances))
          .addService(
              new ContentAddressableStorageService(
                  instances,
                  /* deadlineAfter=*/ 1,
                  TimeUnit.DAYS,
                  /* requestLogLevel=*/ Level.INFO))
          .addService(new ByteStreamService(instances, /* writeDeadlineAfter=*/ 1, TimeUnit.DAYS))
          .addService(
              new ExecutionService(
                  instances,
                  config.getExecuteKeepaliveAfterSeconds(),
                  TimeUnit.SECONDS,
                  keepaliveScheduler,
                  getMetricsPublisher(config.getMetricsConfig())))
          .addService(new OperationQueueService(instances))
          .addService(new OperationsService(instances))
          .intercept(TransmitStatusRuntimeExceptionInterceptor.instance())
          .intercept(headersInterceptor)
          .build();

  logger.log(Level.INFO, String.format("%s initialized", session));
}