class BaragonRequestWorker implements Runnable {

  @Override
  public void run() {
    workerLastStartAt.set(System.currentTimeMillis());

    try {
      final List<QueuedRequestWithState> queuedRequests = requestManager.getQueuedRequestIds()
          .stream()
          .map(this::hydrateQueuedRequestWithState)
          .filter(Optional::isPresent)
          .map(Optional::get)
          .collect(Collectors.toList());

      List<QueuedRequestWithState> inFlightRequests = queuedRequests.stream()
          .filter((q) -> q.getCurrentState().isInFlight() || hasInProgressAttempt(q))
          .collect(Collectors.toList());

      final Set<String> inProgressServices = inFlightRequests.stream()
          .map((q) -> q.getQueuedRequestId().getServiceId())
          .collect(Collectors.toSet());

      final Set<QueuedRequestWithState> removedForCurrentInFlightRequest = queuedRequests.stream()
          .filter((q) -> inProgressServices.contains(q.getQueuedRequestId().getServiceId()) && !inFlightRequests.contains(q))
          .collect(Collectors.toSet());
      if (!inFlightRequests.isEmpty()) {
        LOG.info("Skipping new updates for services {} due to current in-flight updates", String.join(",", inProgressServices));
      }
      queuedRequests.removeAll(removedForCurrentInFlightRequest);

      // First process results for any requests that were already in-flight
      LOG.debug("Processing {} BaragonRequests which are already in-flight", inFlightRequests.size());
      handleResultStates(handleQueuedRequests(inFlightRequests));
      queuedRequests.removeAll(inFlightRequests);

      int added = inFlightRequests.size();

      while (added < configuration.getWorkerConfiguration().getMaxRequestsPerPoll() && !queuedRequests.isEmpty()) {
        ArrayList<QueuedRequestWithState> nonServiceChanges = new ArrayList<>();
        ArrayList<QueuedRequestWithState> serviceChanges = new ArrayList<>();

        // Build the batches of requests to be sent to agents
        collectRequests(added, queuedRequests, nonServiceChanges, serviceChanges);

        // Now take the list of non-service-change requests,
        // and sort them such that the quicker noValidate / noReload requests come first.
        List<QueuedRequestWithState> hydratedNonServiceChanges = nonServiceChanges.stream()
            .filter((q) -> {
              // Filter here as well since the Set has changed by the second run of the while loop
              if (inProgressServices.contains(q.getQueuedRequestId().getServiceId())) {
                LOG.info("Skipping {} because {} already has an in progress request", q.getQueuedRequestId().getRequestId(), q.getQueuedRequestId().getServiceId());
                return false;
              }
              return true;
            })
            .map(someRequest -> new MaybeAdjustedRequest(someRequest, false))
            .map(this::setNoValidateIfRequestRemovesUpstreamsOnly)
            .map(this::preResolveDNS)
            .map(this::saveAdjustedRequest)
            .sorted(queuedRequestComparator())
            .collect(Collectors.toList());

        added += hydratedNonServiceChanges.size();

        // Then send them off.
        LOG.debug("Processing {} BaragonRequests which don't modify a BaragonService", nonServiceChanges.size());
        handleResultStates(handleQueuedRequests(hydratedNonServiceChanges));

        queuedRequests.removeAll(nonServiceChanges);
        inProgressServices.addAll(hydratedNonServiceChanges.stream().map((q) -> q.getQueuedRequestId().getServiceId()).collect(Collectors.toSet()));

        // Now send off the service change requests, after filtering for services already in flight
        List<QueuedRequestWithState> hydratedServiceChanges = serviceChanges.stream()
            .filter((q) -> {
              if (inProgressServices.contains(q.getQueuedRequestId().getServiceId())) {
                LOG.info("Skipping {} because {} already has an in progress request", q.getQueuedRequestId().getRequestId(), q.getQueuedRequestId().getServiceId());
                return false;
              }
              return true;
            })
            .sorted(queuedRequestComparator())
            .collect(Collectors.toList());

        added += hydratedServiceChanges.size();

        LOG.debug("Processing {} BaragonRequests which modify a BaragonService", serviceChanges.size());
        handleResultStates(handleQueuedRequests(hydratedServiceChanges));

        queuedRequests.removeAll(serviceChanges);
        inProgressServices.addAll(hydratedServiceChanges.stream().map((q) -> q.getQueuedRequestId().getServiceId()).collect(Collectors.toSet()));

        // ...and repeat until we've processed up to the limit of requests
      }
    } catch (Exception e) {
      LOG.warn("Caught exception", e);
      exceptionNotifier.notify(e, Collections.emptyMap());
    } finally {
      LOG.debug("Finished poller loop.");
    }
  }

  @Inject public  BaragonRequestWorker(AgentManager agentManager,
                              RequestManager requestManager,
                              BaragonStateDatastore stateDatastore,
                              BaragonExceptionNotifier exceptionNotifier,
                              BaragonConfiguration configuration,
                              EdgeCache edgeCache,
                              UpstreamResolver resolver,
                              @Named(BaragonDataModule.BARAGON_SERVICE_WORKER_LAST_START) AtomicLong workerLastStartAt);

}