package com.hubspot.singularity.data;

import java.util.List;
import java.util.Map;

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

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.inject.Inject;
import com.hubspot.singularity.SingularityDeploy;
import com.hubspot.singularity.SingularityDeployKey;
import com.hubspot.singularity.SingularityPendingTask;
import com.hubspot.singularity.SingularityRequestWithState;
import com.hubspot.singularity.SingularityTaskRequest;

public class TaskRequestManager {

  private final static Logger LOG = LoggerFactory.getLogger(TaskRequestManager.class);
  
  private final DeployManager deployManager;
  private final RequestManager requestManager;
  
  @Inject
  public TaskRequestManager(DeployManager deployManager, RequestManager requestManager) {
    this.deployManager = deployManager;
    this.requestManager = requestManager;
  }

  public List<SingularityTaskRequest> getTaskRequests(List<SingularityPendingTask> tasks) {
    final Map<String, SingularityPendingTask> requestIdToPendingTaskId = Maps.newHashMapWithExpectedSize(tasks.size());
    
    for (SingularityPendingTask task : tasks) {
      requestIdToPendingTaskId.put(task.getPendingTaskId().getRequestId(), task);
    }
    
    final List<SingularityRequestWithState> matchingRequests = requestManager.getRequests(requestIdToPendingTaskId.keySet());
    
    final Map<SingularityPendingTask, SingularityDeployKey> deployKeys = SingularityDeployKey.fromPendingTasks(requestIdToPendingTaskId.values());
    final Map<SingularityDeployKey, SingularityDeploy> matchingDeploys = deployManager.getDeploysForKeys(deployKeys.values());
    
    final List<SingularityTaskRequest> taskRequests = Lists.newArrayListWithCapacity(matchingRequests.size());
    
    for (SingularityRequestWithState request : matchingRequests) {
      SingularityPendingTask task = requestIdToPendingTaskId.get(request.getRequest().getId());
    
      SingularityDeploy foundDeploy = matchingDeploys.get(deployKeys.get(task));
      
      if (foundDeploy == null) {
        LOG.warn("Couldn't find a matching deploy for pending task {}", task);
        continue;
      }
      
      if (!request.getState().isRunnable()) {
        LOG.warn("Request was in state {} for pending task {}", request.getState(), task);
        continue;
      }
      
      taskRequests.add(new SingularityTaskRequest(request.getRequest(), foundDeploy, task));
    }
    
    return taskRequests;
  }
  
}

@@ -173,51 +173,146 @@
 ect.
-Lists;%0Aimport com.google.common.collect.Map
+ArrayListMultimap;%0Aimport com.google.common.collect.Lists;%0Aimport com.google.common.collect.Multimap;%0Aimport com.google.common.collect.Set
 s;%0Ai
@@ -1123,24 +1123,29 @@
 %0A    final M
+ultim
 ap%3CString, S
@@ -1198,38 +1198,31 @@
 d = 
-Maps.newHashMapWithExpectedSiz
+ArrayListMultimap.creat
 e(ta
@@ -1223,32 +1223,35 @@
 ate(tasks.size()
+, 1
 );%0A    %0A    for 
@@ -1766,16 +1766,32 @@
 ForKeys(
+Sets.newHashSet(
 deployKe
@@ -1798,24 +1798,25 @@
 ys.values())
+)
 ;%0A    %0A    f
@@ -1995,24 +1995,29 @@
 ts) %7B%0A      
+for (
 SingularityP
@@ -2032,17 +2032,17 @@
 sk task 
-=
+:
  request
@@ -2096,21 +2096,20 @@
 d())
-;
+) %7B
 %0A    
-%0A
     
-  
 Sing
@@ -2181,25 +2181,29 @@
 sk));%0A      
-%0A
+  %0A  
       if (fo
@@ -2223,32 +2223,34 @@
 null) %7B%0A        
+  
 LOG.warn(%22Couldn
@@ -2304,32 +2304,34 @@
  task);%0A        
+  
 continue;%0A      
@@ -2326,32 +2326,34 @@
 inue;%0A      
+ 
+ 
 %7D%0A      
 %0A      if (!
@@ -2336,25 +2336,29 @@
     %7D%0A      
-%0A
+  %0A  
       if (!r
@@ -2387,24 +2387,26 @@
 nnable()) %7B%0A
+  
         LOG.
@@ -2484,32 +2484,34 @@
  task);%0A        
+  
 continue;%0A      
@@ -2510,24 +2510,26 @@
 ;%0A      
+  
 %7D%0A      
 %0A      t
@@ -2516,25 +2516,29 @@
     %7D%0A      
+ 
-%0A
+ %0A  
       taskRe
@@ -2614,24 +2614,32 @@
 oy, task));%0A
+      %7D%0A
     %7D%0A    %0A 

fix bug with multiple instances