package uk.sky.cirrus.locking;

import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.DriverException;
import com.google.common.util.concurrent.Uninterruptibles;
import uk.sky.cirrus.locking.exception.CannotAcquireLockException;
import uk.sky.cirrus.locking.exception.CannotReleaseLockException;

import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import static com.datastax.driver.core.ConsistencyLevel.ALL;

public class Lock {

    private static final UUID CLIENT = UUID.randomUUID();

    private final String name;
    private final Session session;

    private Lock(String name, Session session) {
        this.name = name;
        this.session = session;
    }

    public static Lock acquire(LockConfig lockConfig, String keyspace, Session session) {

        Duration pollingInterval = lockConfig.getPollingInterval();
        Duration timeout = lockConfig.getTimeout();

        String name = keyspace + ".schema_migration";
        Statement query = new SimpleStatement("INSERT INTO locks.locks (name, client) VALUES (?, ?) IF NOT EXISTS", name, CLIENT)
                .setConsistencyLevel(ALL);

        long startTime = System.currentTimeMillis();

        while(true) {

            ResultSet resultSet;
            try {
                resultSet = session.execute(query);
            } catch (DriverException e) {
                throw new CannotAcquireLockException("Query failed to execute", e);
            }

            Row lock = resultSet.one();
            boolean lockAcquired = lock.getBool("[applied]");

            if (lockAcquired) {
                return new Lock(name, session);
            } else {
                Uninterruptibles.sleepUninterruptibly(pollingInterval.toMillis(), TimeUnit.MILLISECONDS);

                if (timedOut(timeout, startTime)) {
                    throw new CannotAcquireLockException("Lock currently in use by client: " + lock.getUUID("client"));
                }
            }
        }

    }

    private static boolean timedOut(Duration timeout, long startTime) {
        long currentDuration = System.currentTimeMillis() - startTime;
        return currentDuration >= timeout.toMillis();
    }

    public void release() {
        Statement query = new SimpleStatement("DELETE FROM locks.locks WHERE name = ?", name)
                .setConsistencyLevel(ALL);

        try {
            session.execute(query);
        } catch (DriverException e) {
            throw new CannotReleaseLockException("Query failed to execute", e);
        }
    }
}

@@ -179,16 +179,73 @@
 tibles;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 import u
@@ -299,24 +299,24 @@
 kException;%0A
-
 import uk.sk
@@ -539,24 +539,99 @@
 ass Lock %7B%0A%0A
+    private static final Logger log = LoggerFactory.getLogger(Lock.class);%0A
     private 
@@ -1512,32 +1512,104 @@
 rException e) %7B%0A
+                log.warn(%22Query to acquire lock failed to execute%22, e);%0A
                 
@@ -1822,24 +1822,82 @@
 Acquired) %7B%0A
+                log.debug(%22Lock acquired for: %7B%7D%22, name);%0A
             
@@ -1953,16 +1953,160 @@
  else %7B%0A
+                UUID clientWithLock = lock.getUUID(%22client%22);%0A                log.debug(%22Lock currently in use by client: %7B%7D%22, clientWithLock);%0A
         
@@ -2252,24 +2252,128 @@
 artTime)) %7B%0A
+                    log.warn(%22Unable to acquire lock, currently in use by client: %7B%7D%22, clientWithLock);%0A
             
@@ -2455,38 +2455,30 @@
 nt: %22 + 
-lock.getUUID(%22client%22)
+clientWithLock
 );%0A     
@@ -2954,37 +2954,159 @@
     
-%7D catch (DriverException e) %7B
+    log.debug(%22Lock released for: %7B%7D%22, name);%0A        %7D catch (DriverException e) %7B%0A            log.warn(%22Query to release lock failed to execute%22, e);
 %0A   

Add logging to locking