package io.cattle.platform.engine.manager.impl;

import com.google.common.util.concurrent.ListenableFuture;
import io.cattle.platform.async.utils.AsyncUtils;
import io.cattle.platform.engine.manager.LoopFactory;
import io.cattle.platform.engine.manager.LoopManager;
import io.cattle.platform.engine.model.Loop;
import io.cattle.platform.object.ObjectManager;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;

public class LoopManagerImpl implements LoopManager {

    Map<String, Map<String, LoopWrapper>> loops = new HashMap<>();
    LoopFactory factory;
    ExecutorService executor;
    ObjectManager objectManager;
    ScheduledExecutorService executorService;

    public LoopManagerImpl(LoopFactory factory, ExecutorService executor, ObjectManager objectManager, ScheduledExecutorService executorService) {
        super();
        this.factory = factory;
        this.executor = executor;
        this.objectManager = objectManager;
        this.executorService = executorService;
    }

    @Override
    public ListenableFuture<?> kick(String name, String type, Long id, Object input) {
        if (id == null) {
            return AsyncUtils.done();
        }
        String resourceKey = String.format("%s:%s", type, id);
        LoopWrapper loop = getLoop(name, resourceKey);
        if (loop == null) {
            loop = buildLoop(name, resourceKey, type, id);
        }
        if (loop == null) {
            return AsyncUtils.done();
        }

        return loop.kick(input);
    }

    protected synchronized LoopWrapper buildLoop(String name, String resourceKey, String type, Long id) {
        LoopWrapper check = getLoop(name, resourceKey);
        if (check != null) {
            return check;
        }

        Loop inner = factory.buildLoop(name, type, id);
        if (inner == null) {
            return null;
        }
        LoopWrapper loop = new LoopWrapper(name, inner, executor, executorService);
        Map<String, LoopWrapper> loops = this.loops.get(name);
        if (loops == null) {
            loops = new HashMap<>();
            this.loops.put(name, loops);
        }
        loops.put(resourceKey, loop);
        return loop;
    }

    protected synchronized LoopWrapper getLoop(String name, String resourceKey) {
        Map<String, LoopWrapper> loops = this.loops.get(name);
        if (loops == null) {
            return null;
        }
        return loops.get(resourceKey);
    }

    @Override
    public ListenableFuture<?> kick(String name, Class<?> type, Long id, Object input) {
        return kick(name, objectManager.getType(type), id, input);
    }

}

@@ -42,16 +42,99 @@
 .impl;%0A%0A
+import com.google.common.cache.Cache;%0Aimport com.google.common.cache.CacheBuilder;%0A
 import c
@@ -439,55 +439,112 @@
 er;%0A
-%0A
 import 
-java.util.HashMap;%0Aimport java.util.Map
+org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport java.util.concurrent.ExecutionException
 ;%0Aim
@@ -704,71 +704,91 @@
     
-Map%3CString, Map%3CString, LoopWrapper%3E%3E loops = new HashMap%3C%3E(
+private static final Logger log = LoggerFactory.getLogger(LoopManagerImpl.class
 );%0A
+%0A
     
@@ -916,16 +916,90 @@
 Service;
+%0A    Cache%3CString, LoopWrapper%3E loops = CacheBuilder.newBuilder().build();
 %0A%0A    pu
@@ -1500,222 +1500,60 @@
   %7D%0A
-        String resourceKey = String.format(%22%25s:%25s%22, type, id);%0A        LoopWrapper loop = getLoop(name, resourceKey);%0A        if (loop == null) %7B%0A            loop = buildLoop(name, resourceKey, type, id);%0A        %7D
+%0A        LoopWrapper loop = getLoop(name, type, id);
 %0A   
@@ -1676,37 +1676,24 @@
   protected 
-synchronized 
 LoopWrapper 
@@ -1692,21 +1692,19 @@
 Wrapper 
-build
+get
 Loop(Str
@@ -1716,28 +1716,8 @@
 ame,
- String resourceKey,
  Str
@@ -1749,110 +1749,124 @@
     
-LoopWrapper check = getLoop(name, resourceKey);%0A        if (check != null) %7B%0A            return check;
+String key = String.format(%22%25s/%25s:%25s%22, name, type, id);%0A        try %7B%0A            return this.loops.get(key, () -%3E %7B
 %0A   
@@ -1862,35 +1862,32 @@
 () -%3E %7B%0A        
-%7D%0A%0A
         Loop inn
@@ -1918,32 +1918,40 @@
 ame, type, id);%0A
+        
         if (inne
@@ -1967,32 +1967,40 @@
 ) %7B%0A            
+        
 return null;%0A   
@@ -1988,32 +1988,40 @@
    return null;%0A
+        
         %7D%0A      
@@ -2022,34 +2022,31 @@
         
-LoopWrapper loop =
+        return 
  new Loo
@@ -2107,493 +2107,153 @@
     
-Map%3CString, LoopWrapper%3E loops = this.loops.get(name);%0A        if (loops == null) %7B%0A            loops = new HashMap%3C%3E();%0A            this.loops.put(name, loops);%0A        %7D%0A        loops.put(resourceKey, loop);%0A        return loop;%0A    %7D%0A%0A    protected synchronized LoopWrapper getLoop(String name, String resourceKey) %7B%0A        Map%3CString, LoopWrapper%3E loops = this.loops.get(name);%0A        if (loops == null) %7B%0A            return null;%0A        %7D%0A        return loops.get(resourceKey);
+    %7D);%0A        %7D catch (ExecutionException e) %7B%0A            log.error(%22Failed to build loop for %7B%7D%22, key, e);%0A            return null;%0A        %7D
 %0A   

Fix deadlock. Also introduce cache for eviction later