package org.safehaus.subutai.impl.storm.handler;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.safehaus.subutai.api.lxcmanager.LxcCreateException;
import org.safehaus.subutai.impl.storm.StormImpl;
import org.safehaus.subutai.shared.operation.ProductOperationState;
import org.safehaus.subutai.shared.operation.ProductOperationView;
import org.safehaus.subutai.shared.protocol.Agent;

class Helper {

    final StormImpl manager;

    public Helper(StormImpl manager) {
        this.manager = manager;
    }

    Set<Agent> createContainers(int count) throws LxcCreateException {
        Map<Agent, Set<Agent>> lxcs = manager.getLxcManager().createLxcs(count);
        Set<Agent> res = new HashSet<>();
        for(Set<Agent> s : lxcs.values()) res.addAll(s);
        return res;
    }

    Agent createContainer() throws LxcCreateException {
        Set<Agent> s = createContainers(1);
        return s.size() > 0 ? s.iterator().next() : null;
    }

    boolean installZookeeper(Agent agent) {
        UUID id = manager.getZookeeperManager().install(agent.getHostname());
        CountDownLatch latch = new CountDownLatch(1);
        new Thread(new OperationStatusChecker(latch, id)).start();
        try {
            latch.await(2, TimeUnit.MINUTES);
        } catch(InterruptedException ex) {
            return false;
        }
        return true;
    }

    boolean startZookeeper(Agent agent) {
        UUID id = manager.getZookeeperManager().start(agent.getHostname());
        CountDownLatch latch = new CountDownLatch(1);
        new Thread(new OperationStatusChecker(latch, id)).start();
        try {
            latch.await(1, TimeUnit.MINUTES);
        } catch(InterruptedException ex) {
            return false;
        }
        return true;
    }

    class OperationStatusChecker implements Runnable {

        final CountDownLatch latch;
        final UUID trackerId;

        public OperationStatusChecker(CountDownLatch latch, UUID trackerId) {
            this.latch = latch;
            this.trackerId = trackerId;
        }

        @Override
        public void run() {
            String zk = org.safehaus.subutai.api.zookeeper.Config.PRODUCT_KEY;
            ProductOperationView p = null;
            while(p == null || p.getState() == ProductOperationState.RUNNING) {
                p = manager.getTracker().getProductOperation(zk, trackerId);
                try {
                    Thread.sleep(200);
                } catch(InterruptedException ex) {
                    break;
                }
            }
            latch.countDown();
        }

    }
}

@@ -1146,405 +1146,495 @@
     
-CountDownLatch latch = new CountDownLatch(1);%0A        new Thread(new OperationStatusChecker(latch, id)).start();%0A        try %7B%0A            latch.await(2, TimeUnit.MINUTES);%0A        %7D catch(InterruptedException ex) %7B%0A            return false;%0A        %7D%0A        return true;%0A    %7D%0A%0A    boolean startZookeeper(Agent agent) %7B%0A        UUID id = manager.getZookeeperManager().start(agent.getHostname());
+String src = org.safehaus.subutai.api.zookeeper.Config.PRODUCT_KEY;%0A        return watchOperation(src, id, 2, TimeUnit.MINUTES);%0A    %7D%0A%0A    boolean startZookeeper(Agent agent) %7B%0A        UUID id = manager.getZookeeperManager().start(agent.getHostname());%0A        String src = org.safehaus.subutai.api.zookeeper.Config.PRODUCT_KEY;%0A        return watchOperation(src, id, 1, TimeUnit.MINUTES);%0A    %7D%0A%0A    private boolean watchOperation(String source, UUID id, long timeout, TimeUnit unit) %7B
 %0A   
@@ -1736,16 +1736,24 @@
 r(latch,
+ source,
  id)).st
@@ -1781,24 +1781,31 @@
             
+return 
 latch.await(
@@ -1804,35 +1804,29 @@
 h.await(
-1, TimeUnit.MINUTES
+timeout, unit
 );%0A     
@@ -1903,29 +1903,8 @@
   %7D%0A
-        return true;%0A
     
@@ -1990,32 +1990,61 @@
 ownLatch latch;%0A
+        final String source;%0A
         final UU
@@ -2117,16 +2117,31 @@
 h latch,
+ String source,
  UUID tr
@@ -2175,32 +2175,66 @@
 .latch = latch;%0A
+            this.source = source;%0A
             this
@@ -2330,75 +2330,51 @@
     
-String zk = org.safehaus.subutai.api.zookeeper.Config.PRODUCT_KEY;%0A
+if(source != null && trackerId != null) %7B%0A 
     
@@ -2381,16 +2381,19 @@
         
+   
 ProductO
@@ -2411,24 +2411,28 @@
 w p = null;%0A
+    
             
@@ -2515,16 +2515,20 @@
         
+    
 p = mana
@@ -2568,10 +2568,14 @@
 ion(
-zk
+source
 , tr
@@ -2592,32 +2592,36 @@
                 
+    
 try %7B%0A          
@@ -2630,16 +2630,20 @@
         
+  
+  
 Thread.s
@@ -2653,16 +2653,20 @@
 p(200);%0A
+    
         
@@ -2728,16 +2728,20 @@
         
+    
 break;%0A 
@@ -2747,34 +2747,42 @@
                 
+  
-%7D%0A
+  %7D%0A    
             %7D%0A  
@@ -2771,32 +2771,36 @@
               %7D%0A
+    
             latc
@@ -2810,24 +2810,38 @@
 ountDown();%0A
+            %7D%0A
         %7D%0A%0A 

refactor Helper class