package com.hubspot.singularity.scheduler;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

import javax.inject.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.name.Named;
import com.hubspot.mesos.JavaUtils;
import com.hubspot.singularity.config.SingularityConfiguration;
import com.hubspot.singularity.mesos.SingularityMesosModule;

@Singleton
public class SingularitySchedulerPoller extends SingularityLeaderOnlyPoller {

  private static final Logger LOG = LoggerFactory.getLogger(SingularitySchedulerPoller.class);

  private final SingularityScheduler scheduler;
  private final Provider<SingularitySchedulerStateCache> stateCacheProvider;

  @Inject
  SingularitySchedulerPoller(Provider<SingularitySchedulerStateCache> stateCacheProvider, SingularityScheduler scheduler,
      SingularityConfiguration configuration, @Named(SingularityMesosModule.SCHEDULER_LOCK_NAME) final Lock lock) {
    super(configuration.getCheckSchedulerEverySeconds(), TimeUnit.SECONDS, lock);

    this.stateCacheProvider = stateCacheProvider;
    this.scheduler = scheduler;
  }

  @Override
  public void runActionOnPoll() {
    final long start = System.currentTimeMillis();

    final SingularitySchedulerStateCache stateCache = stateCacheProvider.get();

    scheduler.checkForDecomissions(stateCache);
    scheduler.drainPendingQueue(stateCache);

    LOG.debug("Processed decomissions and pending queue in {}", JavaUtils.duration(start));
  }

}

@@ -37,16 +37,39 @@
 duler;%0A%0A
+import java.util.List;%0A
 import j
@@ -171,16 +171,97 @@
 leton;%0A%0A
+import org.apache.mesos.Protos.OfferID;%0Aimport org.apache.mesos.SchedulerDriver;%0A
 import o
@@ -332,21 +332,67 @@
 gle.
-inject.Inject
+common.base.Optional;%0Aimport com.google.common.collect.Sets
 ;%0Aim
@@ -414,24 +414,22 @@
 .inject.
-Provider
+Inject
 ;%0Aimport
@@ -596,38 +596,279 @@
 y.mesos.
-SingularityMesosModule
+OfferCache;%0Aimport com.hubspot.singularity.mesos.SchedulerDriverSupplier;%0Aimport com.hubspot.singularity.mesos.SingularityMesosModule;%0Aimport com.hubspot.singularity.mesos.SingularityMesosOfferScheduler;%0Aimport com.hubspot.singularity.mesos.SingularityOfferHolder
 ;%0A%0A@Sing
@@ -1068,36 +1068,92 @@
 nal 
-SingularityScheduler schedul
+OfferCache offerCache;%0A  private final SchedulerDriverSupplier schedulerDriverSuppli
 er;%0A
@@ -1168,25 +1168,16 @@
 e final 
-Provider%3C
 Singular
@@ -1183,45 +1183,40 @@
 rity
-SchedulerStateCache%3E stateCacheProvid
+MesosOfferScheduler offerSchedul
 er;%0A
@@ -1259,17 +1259,8 @@
 ler(
-Provider%3C
 Sing
@@ -1270,77 +1270,112 @@
 rity
-SchedulerStateCache%3E stateCacheProvider, SingularityScheduler schedul
+MesosOfferScheduler offerScheduler, OfferCache offerCache, SchedulerDriverSupplier schedulerDriverSuppli
 er,%0A
@@ -1586,77 +1586,131 @@
 his.
-stateCacheProvider = stateCacheProvider;%0A    this.scheduler = schedul
+offerCache = offerCache;%0A    this.offerScheduler = offerScheduler;%0A    this.schedulerDriverSupplier = schedulerDriverSuppli
 er;%0A
@@ -1816,22 +1816,21 @@
 );%0A%0A    
-final 
+List%3C
 Singular
@@ -1836,224 +1836,726 @@
 rity
-SchedulerStateCache stateCache = stateCacheProvider.get();%0A%0A    scheduler.checkForDecomissions(stateCache);%0A    scheduler.drainPendingQueue(stateCache);%0A%0A    LOG.debug(%22Processed decomissions and pending queue in %7B%7D%22
+OfferHolder%3E offerHolders = offerScheduler.checkOffers(offerCache.getCachedOffers(), Sets.%3COfferID%3E newHashSet());%0A%0A    if (offerHolders.isEmpty()) %7B%0A      return;%0A    %7D%0A%0A    Optional%3CSchedulerDriver%3E driver = schedulerDriverSupplier.get();%0A%0A    if (!driver.isPresent()) %7B%0A      LOG.error(%22No driver present, can't accept cached offers%22);%0A      return;%0A    %7D%0A%0A    int acceptedOffers = 0;%0A%0A    for (SingularityOfferHolder offerHolder : offerHolders) %7B%0A      if (!offerHolder.getAcceptedTasks().isEmpty()) %7B%0A        offerHolder.launchTasks(driver.get());%0A        acceptedOffers++;%0A        offerCache.useOffer(offerHolder.getOffer().getId());%0A      %7D%0A    %7D%0A%0A    LOG.debug(%22Accepted %7B%7D cached offers in %7B%7D%22, acceptedOffers
 , Ja

check cached offers on scheduler poller