/**
 * Copyright 2015 Palantir Technologies
 * <p>
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://opensource.org/licenses/BSD-3-Clause
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.palantir.atlasdb.keyvalue.cassandra;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.thrift.TException;
import org.joda.time.Duration;

import com.jayway.awaitility.Awaitility;
import com.jayway.awaitility.core.ConditionTimeoutException;
import com.palantir.atlasdb.cassandra.CassandraKeyValueServiceConfig;
import com.palantir.common.base.FunctionCheckedException;

/**
 * Utilities for ETE tests
 * Created by aloro on 12/04/2016.
 */
class CassandraTestTools {
    private CassandraTestTools() {
        // Empty constructor for utility class
    }

    static void waitTillServiceIsUp(String host, int port, Duration timeout) {
        try {
            Awaitility.await()
                    .pollInterval(50, TimeUnit.MILLISECONDS)
                    .atMost(timeout.getMillis(), TimeUnit.MILLISECONDS)
                    .until(isPortListening(host, port));
        } catch (ConditionTimeoutException e) {
            throw new IllegalStateException("Timeout for port " + port + " on host " + host + ".");
        }
    }

    private static Callable<Boolean> isPortListening(String host, int port) {
        return () -> {
            try (Socket socket = new Socket()) {
                socket.connect(new InetSocketAddress(host, port), 500);
                return true;
            } catch (IOException e) {
                return false;
            }
        };
    }

    static Future async(ExecutorService executorService, Runnable callable) {
        return executorService.submit(callable);
    }

    static void assertThatFutureDidNotSucceedYet(Future future) throws InterruptedException {
        if (future.isDone()) {
            try {
                future.get();
                throw new AssertionError("Future task should have failed but finished successfully");
            } catch (ExecutionException e) {
                // if execution is done, we expect it to have failed
            }
        }
    }

    static void dropTables(final CassandraKeyValueServiceConfig config) throws TException {
        CassandraClientPool clientPool = new CassandraClientPool(config);
        clientPool.run(new FunctionCheckedException<Cassandra.Client, Void, TException>() {
            @Override
            public Void apply(Cassandra.Client client) throws TException {
                client.system_drop_keyspace(config.keyspace());
                return null;
            }
        });
    }
}

@@ -1016,39 +1016,121 @@
 ra.t
-ransport.messages.ResultMessage
+hrift.CfDef;%0Aimport org.apache.cassandra.thrift.InvalidRequestException;%0Aimport org.apache.cassandra.thrift.KsDef
 ;%0Aim
@@ -1196,16 +1196,64 @@
 ation;%0A%0A
+import com.google.common.collect.ImmutableList;%0A
 import c
@@ -1440,44 +1440,65 @@
 tir.
-common.base.FunctionCheckedException
+atlasdb.cassandra.ImmutableCassandraKeyValueServiceConfig
 ;%0A%0A/
@@ -3090,13 +3090,23 @@
 drop
-Table
+KeyspaceIfExist
 s(fi
@@ -3160,32 +3160,50 @@
 ws TException %7B%0A
+        try %7B%0A    
         Cassandr
@@ -3272,42 +3272,219 @@
     
-clientPool.run(new FunctionChecked
+    String keyspace = config.keyspace();%0A            clientPool.run(client -%3E %7B%0A                client.system_drop_keyspace(keyspace);%0A                return null;%0A            %7D);%0A        %7D catch (InvalidRequest
 Exce
@@ -3488,17 +3488,19 @@
 xception
-%3C
+ %7C 
 Cassandr
@@ -3504,24 +3504,42 @@
 ndra
-.Client, Void, T
+ClientFactory.ClientCreationFailed
 Exce
@@ -3547,10 +3547,10 @@
 tion
-%3E(
+ e
 ) %7B%0A
@@ -3565,209 +3565,1368 @@
     
-@Override%0A            public Void apply(Cassandra.Client client) throws TException %7B%0A                client.system_drop_keyspace(config.keyspace());%0A                return null;%0A        
+// This happens when the keyspace didn't exist in the first place. In this case, our work is done.%0A            System.out.println(%22Didn't drop keyspace %22 + config.keyspace() + %22, likely because it didn't exist%22);%0A            System.out.println(%22Message: %22 + e.getMessage());%0A            System.out.println(%22Cause: %22 + e.getCause().getMessage());%0A        %7D%0A    %7D%0A%0A    static void createKeyspace(ImmutableCassandraKeyValueServiceConfig config) throws TException %7B%0A        // Shamelessly copied from CassandraVerifier. TODO does it make sense to deduplicate this code?%0A        InetSocketAddress host = config.servers().iterator().next();%0A        Cassandra.Client client = CassandraClientFactory.getClientInternal(host, config.credentials(),%0A                config.ssl(), config.socketTimeoutMillis(), config.socketQueryTimeoutMillis());%0A        KsDef ks = new KsDef(config.keyspace(), CassandraConstants.NETWORK_STRATEGY, ImmutableList.%3CCfDef%3Eof());%0A        CassandraVerifier.checkAndSetReplicationFactor(client, ks, true, config.replicationFactor(), config.safetyDisabled());%0A        ks.setDurable_writes(true);%0A        System.out.println(%22Creating keyspace: %22 + config.keyspace());%0A        client.system_add_keyspace(ks);%0A
     
-%7D%0A
     
-    %7D
+CassandraKeyValueServices.waitForSchemaVersions(client, %22(adding the initial empty keyspace)%22, config.schemaMutationTimeoutMillis()
 );%0A 

allow tests to create keyspace without using CKVS.create