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

}

class BaragonRequestWorkerTest  {

  private static final String TEST_LB_GROUP;
  @Inject RequestManager requestManager;
  @Inject BaragonRequestWorker requestWorker;

  @Test
  public void testFailedRetriedRequestsAreConsideredInFlight() throws Exception {

    String agentUrl = "http://agent2";
    startAgent(agentUrl, TEST_LB_GROUP);
    requestManager.enqueueRequest(createBaseRequest("request1", "service1", ImmutableSet.of(TEST_LB_GROUP)).build());
    requestWorker.run(); // move from pending -> send apply
    requestWorker.run(); // actually send
    Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size());
    testAgentManager.completeRequestWithFailure(agentUrl, agentUrl, "request1", AgentRequestType.APPLY);
    requestWorker.run(); // see the failed response, move to retry
    Assertions.assertEquals(InternalRequestStates.SEND_APPLY_REQUESTS, requestManager.getRequestState("request1").get());

    // second request for same service should not send a request batch
    requestManager.enqueueRequest(createBaseRequest("request2", "service1", ImmutableSet.of(TEST_LB_GROUP)).build());
    Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size());
    Assertions.assertEquals("request1", testAgentManager.getRecentBatches().get(agentUrl).get(0).getRequestId());
  }

}