package org.apache.mesos.logstash.state;

import org.apache.log4j.Logger;
import org.apache.mesos.Protos.FrameworkID;
import org.apache.mesos.Protos.TaskInfo;
import org.apache.mesos.Protos.TaskState;
import org.apache.mesos.Protos.TaskStatus;

import java.io.IOException;
import java.security.InvalidParameterException;

/**
 * Status of task. This is necessary because the raw TaskInfo packet doesn't contain the frameworkID or a link to
 * the respective TaskStatus packet.
 */
public class LSTaskStatus {
    private static final Logger LOGGER = Logger.getLogger(TaskStatus.class);
    public static final String STATE_KEY = "state";
    public static final String DEFAULT_STATUS_NO_MESSAGE_SET = "Default status. No message set.";
    private final SerializableState state;
    private final FrameworkID frameworkID;

    private final TaskInfo taskInfo;

    private final StatePath statePath;
    public LSTaskStatus(SerializableState state, FrameworkID frameworkID, TaskInfo taskInfo) {
        if (state == null) {
            throw new InvalidParameterException("State cannot be null");
        } else if (frameworkID == null || frameworkID.getValue().isEmpty()) {
            throw new InvalidParameterException("FrameworkID cannot be null or empty");
        }
        this.state = state;
        this.frameworkID = frameworkID;
        this.taskInfo = taskInfo;
        statePath = new StatePath(state);
    }

    public void setStatus(TaskStatus status) throws IllegalStateException {
        try {
            LOGGER.debug("Writing task status to zk: [" + status.getState() + "] " + status.getTaskId().getValue());
            statePath.mkdir(getKey());
            state.set(getKey(), status);
        } catch (IOException e) {
            throw new IllegalStateException("Unable to write task status to zookeeper", e);
        }
    }

    public TaskStatus getStatus() throws IllegalStateException {
        try {
            return state.get(getKey());
        } catch (IOException e) {
            throw new IllegalStateException("Unable to get task status from zookeeper", e);
        }
    }

    public TaskStatus getDefaultStatus() {
        return TaskStatus.newBuilder()
                    .setState(TaskState.TASK_STARTING)
                    .setTaskId(taskInfo.getTaskId())
                    .setExecutorId(taskInfo.getExecutor().getExecutorId())
                    .setMessage(DEFAULT_STATUS_NO_MESSAGE_SET)
                    .build();
    }

    public TaskInfo getTaskInfo() {
        return taskInfo;
    }

    @Override
    public String toString() {
        String retVal;
        try {
            retVal = getKey() + ": [" + getStatus().getState() + "] " +  getStatus().getMessage();
        } catch (Exception e) {
            retVal = getKey() + ": Unable to get message";
        }
        return retVal;
    }

    private String getKey() {
        return frameworkID.getValue() + "/" + STATE_KEY + "/" + taskInfo.getTaskId().getValue();
    }

    public boolean isRunning(){
        TaskState state = getStatus().getState();
        return  TaskState.TASK_RUNNING.equals(state);
    }


    public boolean taskInTerminalState() {
        TaskState state = getStatus().getState();
        return StateUtil.isTerminalState(state);
    }

    public void destroy() {
        try {
            state.delete(getKey());
        } catch (IOException e) {
            LOGGER.error("Could not destroy Task in ZK.", e);
        }
    }
}

@@ -729,16 +729,83 @@
  set.%22;%0A
+    private static final String FINGERPRINT_KEY = %22configuration%22;%0A
     priv
@@ -1476,32 +1476,802 @@
 (state);%0A    %7D%0A%0A
+    public void setConfigurationFingerprint(String fingerprint) %7B%0A        try %7B%0A            LOGGER.debug(%22Writing configuration fingerprint to zk: %5B%22 + fingerprint + %22%5D %22 + taskInfo.getTaskId().getValue());%0A            statePath.mkdir(getKey(FINGERPRINT_KEY));%0A            state.set(getKey(FINGERPRINT_KEY), fingerprint);%0A        %7D catch (IOException e) %7B%0A            throw new IllegalStateException(%22Unable to write configuration fingerprint to zookeeper%22, e);%0A        %7D%0A    %7D%0A%0A    public String getConfigurationFingerprint() throws IllegalStateException %7B%0A        try %7B%0A            return state.get(getKey(FINGERPRINT_KEY));%0A        %7D catch (IOException e) %7B%0A            throw new IllegalStateException(%22Unable to get task status from zookeeper%22, e);%0A        %7D%0A    %7D%0A%0A
     public void 
@@ -2488,32 +2488,41 @@
 th.mkdir(getKey(
+STATE_KEY
 ));%0A            
@@ -2538,16 +2538,25 @@
 (getKey(
+STATE_KEY
 ), statu
@@ -2809,32 +2809,41 @@
 tate.get(getKey(
+STATE_KEY
 ));%0A        %7D ca
@@ -3512,24 +3512,33 @@
 al = getKey(
+STATE_KEY
 ) + %22: %5B%22 + 
@@ -3656,16 +3656,25 @@
  getKey(
+STATE_KEY
 ) + %22: U
@@ -3762,16 +3762,27 @@
  getKey(
+String path
 ) %7B%0A    
@@ -3823,25 +3823,20 @@
 + %22/%22 + 
-STATE_KEY
+path
  + %22/%22 +
@@ -4193,24 +4193,150 @@
 destroy() %7B%0A
+        safeDeleteKey(STATE_KEY);%0A        safeDeleteKey(FINGERPRINT_KEY);%0A    %7D%0A%0A    private void safeDeleteKey(String key) %7B%0A
         try 
@@ -4369,16 +4369,19 @@
 (getKey(
+key
 ));%0A    

Implement functionality for storing configuration finger prints in zookeeper