package com.cisco.oss.foundation.cluster.masterslave.consul;

import com.cisco.oss.foundation.cluster.masterslave.MastershipElector;
import com.cisco.oss.foundation.cluster.utils.MasterSlaveConfigurationUtil;
import com.google.common.base.Optional;
import com.google.common.net.HostAndPort;
import com.orbitz.consul.Consul;
import com.orbitz.consul.ConsulException;
import com.orbitz.consul.model.session.ImmutableSession;
import com.orbitz.consul.model.session.Session;
import com.orbitz.consul.model.session.SessionCreatedResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Consul implementation for logic for electing new masters.
 * Created by Yair Ogen (yaogen) on 15/02/2016.
 */
public class ConsulMastershipElector implements MastershipElector {

    public static final String ACTIVE_DATACENTER = "activeDatacenter";
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsulMastershipElector.class);
    private static final int CONSUL_RETRY_DELAY = 10;
    private Consul consul = null;
    private String activeVersionKey = "";
    private String sessionId = "";
    private String mastershipKey = "";
    public final AtomicBoolean IS_CONSUL_UP = new AtomicBoolean(false);


    @Override
    public void init(String id, String jobName) {
        this.mastershipKey = id;
        this.activeVersionKey = MasterSlaveConfigurationUtil.COMPONENT_NAME+"-version";
        HostAndPort consulHostAndPort = MasterSlaveConfigurationUtil.getConsulHostAndPort(jobName);
        try {
            initConsul(consulHostAndPort);
        } catch (ConsulException e) {
            infiniteConnect(consulHostAndPort, jobName);
        }
    }

    private void initConsul(HostAndPort consulHostAndPort) {

        this.consul = Consul.builder().withHostAndPort(consulHostAndPort).build();
        SessionCreatedResponse session = consul.sessionClient().createSession(ImmutableSession.builder().name(MasterSlaveConfigurationUtil.INSTANCE_ID).build());
        this.sessionId = session.getId();
    }

    private void infiniteConnect(HostAndPort consulHostAndPort, String jobName) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while(!IS_CONSUL_UP.get()){
                    try {
                        initConsul(consulHostAndPort);
                        IS_CONSUL_UP.set(true);
                        LOGGER.info("consul reconnect is successful");
                    } catch (Exception e) {
                        LOGGER.warn("consul reconnect failed. retrying in {} seconds. error: {}", CONSUL_RETRY_DELAY, e);
                        try {
                            TimeUnit.SECONDS.sleep(CONSUL_RETRY_DELAY);
                        } catch (InterruptedException e1) {
                            //ignore
                        }
                    }
                }
            }
        },"Infinite-Reconnect-" + jobName).start();
    }

    @Override
    public boolean isReady() {
        try {
            if(consul != null){
                consul.agentClient().ping();
                return true;
            }else{
                return false;
            }
        } catch (Exception e) {
            LOGGER.error("can't ping the agent. error: {}", e);
            return false;
        }
    }

    @Override
    public boolean isActiveVersion(String currentVersion) {
        Optional<String> valueAsString = consul.keyValueClient().getValueAsString(activeVersionKey);
        if(valueAsString.isPresent()){
            return currentVersion.equals(valueAsString.get());
        }
        return true;
    }

    @Override
    public boolean isActiveDataCenter(String currentDataCenter) {
        Optional<String> valueAsString = consul.keyValueClient().getValueAsString(ACTIVE_DATACENTER);
        if(valueAsString.isPresent()){
            return currentDataCenter.equals(valueAsString.get());
        }
        return true;
    }

    @Override
    public boolean isMaster() {
        boolean lockAcquired = consul.keyValueClient().acquireLock(mastershipKey, sessionId);
        return lockAcquired;
    }

    @Override
    public void close() {
        if (consul != null) {
            consul.keyValueClient().releaseLock(mastershipKey, sessionId);
        }
    }
}

@@ -527,16 +527,54 @@
 sponse;%0A
+import org.apache.commons.io.IOUtils;%0A
 import o
@@ -623,16 +623,93 @@
 ctory;%0A%0A
+import javax.ws.rs.InternalServerErrorException;%0Aimport java.io.InputStream;%0A
 import j
@@ -4234,16 +4234,64 @@
 quired =
+ false;%0A        try %7B%0A            lockAcquired =
  consul.
@@ -4338,32 +4338,543 @@
 ey, sessionId);%0A
+        %7D catch (InternalServerErrorException e) %7B%0A            String response = e.getResponse().readEntity(String.class);%0A            if(response.contains(%22invalid session%22))%7B%0A                SessionCreatedResponse session = consul.sessionClient().createSession(ImmutableSession.builder().name(MasterSlaveConfigurationUtil.INSTANCE_ID).build());%0A                this.sessionId = session.getId();%0A            %7D%0A            lockAcquired = consul.keyValueClient().acquireLock(mastershipKey, sessionId);%0A        %7D%0A
         return l

add support for invalid session fix