package org.safehaus.subutai.plugin.flume.impl.handler;

import java.util.*;
import org.safehaus.subutai.api.commandrunner.*;
import org.safehaus.subutai.plugin.flume.api.FlumeConfig;
import org.safehaus.subutai.plugin.flume.impl.CommandType;
import org.safehaus.subutai.plugin.flume.impl.Commands;
import org.safehaus.subutai.plugin.flume.impl.FlumeImpl;
import org.safehaus.subutai.shared.operation.AbstractOperationHandler;
import org.safehaus.subutai.shared.operation.ProductOperation;
import org.safehaus.subutai.shared.protocol.Agent;

public class DestroyNodeHandler extends AbstractOperationHandler<FlumeImpl> {

    private final String hostname;

    public DestroyNodeHandler(FlumeImpl manager, String clusterName, String hostname) {
        super(manager, clusterName);
        this.hostname = hostname;
        this.productOperation = manager.getTracker().createProductOperation(
                FlumeConfig.PRODUCT_KEY, "Remove node from cluster: " + hostname);
    }

    @Override
    public void run() {
        ProductOperation po = productOperation;
        FlumeConfig config = manager.getCluster(clusterName);
        if(config == null) {
            po.addLogFailed("Cluster does not exist: " + clusterName);
            return;
        }

        Agent agent = manager.getAgentManager().getAgentByHostname(hostname);
        if(agent == null) {
            po.addLogFailed("Node is not connected: " + hostname);
            return;
        }
        if(config.getNodes().size() == 1) {
            po.addLogFailed("This is the last node in the cluster. Destroy cluster instead");
            return;
        }

        po.addLog("Uninstalling Flume...");
        Command cmd = manager.getCommandRunner().createCommand(
                new RequestBuilder(Commands.make(CommandType.PURGE)),
                new HashSet<>(Arrays.asList(agent)));
        manager.getCommandRunner().runCommand(cmd);

        if(cmd.hasCompleted()) {
            AgentResult res = cmd.getResults().get(agent.getUuid());
            if(res.getExitCode() != null && res.getExitCode() == 0)
                if(res.getStdOut().contains("Flume is not installed"))
                    po.addLog("Flume is not installed on " + agent.getHostname());
                else
                    po.addLog("Flume removed from " + agent.getHostname());
            else
                po.addLog(String.format("Error on node %s: %s",
                        agent.getHostname(), res.getStdErr()));

            config.getNodes().remove(agent);

            po.addLog("Updating db...");
            if(manager.getDbManager().saveInfo(FlumeConfig.PRODUCT_KEY, config.getClusterName(), config))
                po.addLogDone("Cluster info updated");
            else
                po.addLogFailed("Failed to save cluster info");
        } else {
            po.addLog(cmd.getAllErrors());
            po.addLogFailed("Uninstallation failed");
        }
    }

}

@@ -1454,24 +1454,184 @@
 ;%0A        %7D%0A
+        if(!config.getNodes().contains(agent)) %7B%0A            po.addLogFailed(%22Node does not belong to Flume installation group%22);%0A            return;%0A        %7D%0A
         if(c
@@ -2262,119 +2262,37 @@
     
-if(res.getStdOut().contains(%22Flume is not installed%22))%0A                    po.addLog(%22Flume is not installed on
+po.addLog(%22Flume removed from
  %22 +
@@ -2322,37 +2322,31 @@
             
-    else%0A    
+else %7B%0A
             
@@ -2363,198 +2363,192 @@
 Log(
-%22Flume removed from %22 + agent.getHostname());%0A            else%0A                po.addLog(String.format(%22Error on node %25s: %25s%22,%0A                        agent.getHostname(), res.getStdErr()));
+res.getStdOut());%0A                po.addLog(res.getStdErr());%0A                po.addLogFailed(%22Failed to remove Flume on %22 + agent.getHostname());%0A                return;%0A            %7D
 %0A%0A  

fix destroy node operation