@Provides
@Singleton
GradingRequestPoller gradingRequestPoller(
        @Named("sealtiel") BasicAuthHeader sealtielClientAuthHeader,
        MessageService messageService,
        Provider<GradingWorker> workerFactory,
        LifecycleEnvironment lifecycleEnvironment) {

    ExecutorService executorService = lifecycleEnvironment.executorService("grading-worker-%d")
            .maxThreads(gradingConfig.getNumWorkerThreads())
            .minThreads(gradingConfig.getNumWorkerThreads())
            .build();

    return new GradingRequestPoller(
            executorService,
            sealtielClientAuthHeader,
            messageService,
            workerFactory);
}
 

@Provides
@Singleton
static GradingResponsePoller gradingResponsePoller(
        LifecycleEnvironment lifecycleEnvironment,
        @Named("sealtiel") BasicAuthHeader sealtielClientAuthHeader,
        MessageService messageService,
        GradingResponseProcessor processor) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("grading-response-processor-%d")
                    .maxThreads(10)
                    .minThreads(10)
                    .build();

    return new GradingResponsePoller(sealtielClientAuthHeader, messageService, executorService, processor);
}
 

@Provides
@Singleton
static GradingResponsePoller gradingResponsePoller(
        LifecycleEnvironment lifecycleEnvironment,
        @Named("sealtiel") BasicAuthHeader sealtielClientAuthHeader,
        MessageService messageService,
        GradingResponseProcessor processor) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("grading-response-processor-%d")
                    .maxThreads(10)
                    .minThreads(10)
                    .build();

    return new GradingResponsePoller(sealtielClientAuthHeader, messageService, executorService, processor);
}
 

@Test
public void testNewManagedCurator() {
    ZooKeeperConfiguration config = parse(ImmutableMap.of("retryPolicy",
            ImmutableMap.builder()
                    .put("type", "untilElapsed")
                    .put("maxElapsedTimeMs", 1000)
                    .put("sleepMsBetweenRetries", 50)
                    .build()));

    LifecycleEnvironment env = mock(LifecycleEnvironment.class);
    CuratorFramework curator = config.newManagedCurator(env);

    assertNotNull(curator);
    assertEquals(CuratorFrameworkState.LATENT, curator.getState());
    verify(env).manage(any(ManagedCuratorFramework.class));
}
 

@Provides
@Singleton
SubmissionRegrader submissionRegrader(
        LifecycleEnvironment lifecycleEnvironment,
        SubmissionStore submissionStore,
        SubmissionRegradeProcessor processor) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("submission-regrade-processor-%d")
                    .maxThreads(5)
                    .minThreads(5)
                    .build();

    return new SubmissionRegrader(submissionStore, executorService, processor);
}
 

@Provides
@Singleton
static ItemSubmissionRegrader itemSubmissionRegrader(
        LifecycleEnvironment lifecycleEnvironment,
        ItemSubmissionStore itemSubmissionStore,
        ItemSubmissionRegradeProcessor processor) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("item-submission-regrade-processor-%d")
                    .maxThreads(5)
                    .minThreads(5)
                    .build();

    return new ItemSubmissionRegrader(itemSubmissionStore, executorService, processor);
}
 

@Provides
@Singleton
SubmissionRegrader submissionRegrader(
        LifecycleEnvironment lifecycleEnvironment,
        SubmissionStore submissionStore,
        SubmissionRegradeProcessor processor) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("submission-regrade-processor-%d")
                    .maxThreads(5)
                    .minThreads(5)
                    .build();

    return new SubmissionRegrader(submissionStore, executorService, processor);
}
 

@Provides
@Singleton
static ItemSubmissionRegrader itemSubmissionRegrader(
        LifecycleEnvironment lifecycleEnvironment,
        ItemSubmissionStore itemSubmissionStore,
        ItemSubmissionRegradeProcessor processor) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("item-submission-regrade-processor-%d")
                    .maxThreads(5)
                    .minThreads(5)
                    .build();

    return new ItemSubmissionRegrader(itemSubmissionStore, executorService, processor);
}
 

@Provides
@Singleton
static ContestScoreboardPoller contestScoreboardPoller(
        UnitOfWorkAwareProxyFactory unitOfWorkAwareProxyFactory,
        LifecycleEnvironment lifecycleEnvironment,
        ContestStore contestStore,
        ContestGroupStore contestGroupStore,
        ContestScoreboardUpdater contestScoreboardUpdater,
        ContestGroupScoreboardUpdater contestGroupScoreboardUpdater) {

    ExecutorService executorService =
            lifecycleEnvironment.executorService("contest-scoreboard-updater-%d")
                    .maxThreads(2)
                    .minThreads(2)
                    .build();

    return unitOfWorkAwareProxyFactory.create(
            ContestScoreboardPoller.class,
            new Class<?>[] {
                    ContestStore.class,
                    ContestGroupStore.class,
                    ExecutorService.class,
                    ContestScoreboardUpdater.class,
                    ContestGroupScoreboardUpdater.class},
            new Object[] {
                    contestStore,
                    contestGroupStore,
                    executorService,
                    contestScoreboardUpdater,
                    contestGroupScoreboardUpdater});
}
 

@Provides
public LifecycleEnvironment lifecycleEnvironment() {
    return environment.lifecycle();
}
 

@Inject
public JudgelsScheduler(LifecycleEnvironment lifecycleEnvironment) {
    this.lifecycleEnvironment = lifecycleEnvironment;
}
 

public ExecutorBuilder(LifecycleEnvironment environment)
{
    this.environment = environment;
}
 

public static List<EventSender> build(
    final Environment environment,
    final CommonConfiguration<?> config,
    final MetricRegistry metricRegistry, final String pubsubHealthcheckTopic) {

  final List<EventSender> senders = new ArrayList<>();

  final KafkaClientProvider kafkaClientProvider =
      new KafkaClientProvider(config.getKafkaBrokers());

  final Optional<KafkaProducer<String, byte[]>> kafkaProducer =
      kafkaClientProvider.getDefaultProducer();

  kafkaProducer.ifPresent(producer -> senders.add(new KafkaSender(producer)));

  final LifecycleEnvironment lifecycle = environment.lifecycle();

  if (!config.getPubsubPrefixes().isEmpty()) {
    final PubSub pubsub = PubSubOptions.getDefaultInstance().getService();

    final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
        new ThreadFactoryBuilder()
            .setDaemon(true)
            .setNameFormat("pubsub-healthchecker-%d")
            .build()
    );

    // choose an arbitrary prefix to use in the healthcheck. we assume if we can connect to
    // one we can connect to all
    final String topicToHealthcheck =
        config.getPubsubPrefixes().iterator().next() + pubsubHealthcheckTopic;

    final GooglePubSubSender.DefaultHealthChecker healthchecker =
        new GooglePubSubSender.DefaultHealthChecker(pubsub, topicToHealthcheck, executor,
            Duration.ofMinutes(5));

    metricRegistry.register("pubsub-health", (Gauge<Boolean>) healthchecker::isHealthy);

    for (final String prefix : config.getPubsubPrefixes()) {
      final GooglePubSubSender sender = GooglePubSubSender.create(pubsub, prefix, healthchecker);
      senders.add(sender);
    }

    lifecycle.manage(new ManagedPubSub(pubsub));
  }

  senders.forEach(lifecycle::manage);

  return senders;
}
 

/**
 * Return a managed Curator connection.  This created connection will be wrapped in a
 * {@link ManagedCuratorFramework} and offered to the provided {@link LifecycleEnvironment} parameter.
 */
public CuratorFramework newManagedCurator(LifecycleEnvironment env) {
    CuratorFramework curator = newCurator();
    env.manage(new ManagedCuratorFramework(curator));
    return curator;
}