@Inject
public PubsubEmulatorServer(
    ConfigurationManager configurationManager,
    PublisherService publisher,
    SubscriberService subscriber,
    AdminService admin,
    HealthStatusManager healthStatusManager) {
  this.publisher = publisher;
  this.subscriber = subscriber;
  this.healthStatusManager = healthStatusManager;

  ServerBuilder builder =
      ServerBuilder.forPort(configurationManager.getServer().getPort())
          .addService(publisher)
          .addService(subscriber)
          .addService(admin)
          .addService(healthStatusManager.getHealthService())
          .maxInboundMessageSize(MAX_MESSAGE_SIZE);
  if (configurationManager.getServer().hasSecurity()) {
    builder.useTransportSecurity(
        new File(configurationManager.getServer().getSecurity().getCertificateChainFile()),
        new File(configurationManager.getServer().getSecurity().getPrivateKeyFile()));
  }
  server = builder.build();
}
 

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

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

@Bean
CompositeHealthContributor grpcServices(GrpcServiceDiscoverer grpcServiceDiscoverer,
                                        HealthStatusManager healthStatusManager) {

    Map<String, HealthIndicator> healthIndicators = new LinkedHashMap<>();

    for (GrpcServiceDefinition grpcService : grpcServiceDiscoverer.findGrpcServices()) {
        String serviceName = grpcService.getDefinition().getServiceDescriptor().getName();
        healthIndicators.put(serviceName, new GrpcHealthIndicator(healthStatusManager, serviceName));
    }

    return CompositeHealthContributor.fromMap(healthIndicators);
}
 

@Override
protected void configure() {
  bind(KafkaClientFactory.class).to(DefaultKafkaClientFactory.class);
  bind(Clock.class).toInstance(Clock.systemUTC());

  bind(ConfigurationManager.class);
  bind(SubscriptionManagerFactory.class);
  bind(PublisherService.class);
  bind(SubscriberService.class);
  bind(AdminService.class);
  bind(StatisticsManager.class);
  bind(HealthStatusManager.class);
  bind(PubsubEmulatorServer.class);
}
 

public void start() throws IOException {
  actionCacheRequestCounter.start();
  instances.start();
  server.start();
  healthStatusManager.setStatus(
      HealthStatusManager.SERVICE_NAME_ALL_SERVICES, ServingStatus.SERVING);
}
 

public void stop() {
  synchronized (this) {
    if (stopping) {
      return;
    }
    stopping = true;
  }
  healthStatusManager.setStatus(
      HealthStatusManager.SERVICE_NAME_ALL_SERVICES, ServingStatus.NOT_SERVING);
  try {
    if (server != null) {
      server.shutdown();
    }
    instances.stop();
    server.awaitTermination(10, TimeUnit.SECONDS);
  } catch (InterruptedException e) {
    if (server != null) {
      server.shutdownNow();
    }
  }
  if (!shutdownAndAwaitTermination(keepaliveScheduler, 10, TimeUnit.SECONDS)) {
    logger.log(Level.WARNING, "could not shut down keepalive scheduler");
  }
  if (!actionCacheRequestCounter.stop()) {
    logger.log(Level.WARNING, "count not shut down action cache request counter");
  }
}
 

private void start() throws Exception {
  health = new HealthStatusManager();
  server =
      NettyServerBuilder.forPort(port)
          .addService(new TestServiceImpl(serverId))
          .addService(new XdsUpdateHealthServiceImpl(health))
          .addService(health.getHealthService())
          .addService(ProtoReflectionService.newInstance())
          .build()
          .start();
  health.setStatus("", ServingStatus.SERVING);
}
 

@ConditionalOnMissingBean
@Bean
public HealthStatusManager healthStatusManager() {
    return new HealthStatusManager();
}
 

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

@Bean
public HealthStatusManager healthStatusManager() {
    return new HealthStatusManager();
}
 

@ConditionalOnMissingBean
@Bean
public HealthStatusManager healthStatusManager() {
    return new HealthStatusManager();
}
 

private XdsUpdateHealthServiceImpl(HealthStatusManager health) {
  this.health = health;
}