/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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.datastax.driver.core;

import static com.datastax.driver.core.ConditionChecker.check;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;

import com.datastax.driver.core.Cluster.Builder;
import com.datastax.driver.core.policies.RoundRobinPolicy;
import com.datastax.driver.core.policies.WhiteListPolicy;
import com.google.common.base.Predicate;
import com.google.common.base.Throwables;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Uninterruptibles;
import com.sun.management.OperatingSystemMXBean;
import io.netty.channel.EventLoopGroup;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Level;
import org.scassandra.Scassandra;
import org.scassandra.ScassandraFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.SkipException;

/** A number of static fields/methods handy for tests. */
public abstract class TestUtils {

  public static final String IP_PREFIX;

  static {
    String ip_prefix = System.getProperty("ipprefix");
    if (ip_prefix == null || ip_prefix.isEmpty()) {
      ip_prefix = "127.0.1.";
    }
    IP_PREFIX = ip_prefix;
  }

  private static final Logger logger = LoggerFactory.getLogger(TestUtils.class);

  public static final String CREATE_KEYSPACE_SIMPLE_FORMAT =
      "CREATE KEYSPACE %s WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : %d }";
  public static final String CREATE_KEYSPACE_GENERIC_FORMAT =
      "CREATE KEYSPACE %s WITH replication = { 'class' : '%s', %s }";

  public static final String SELECT_ALL_FORMAT = "SELECT * FROM %s";

  public static final int TEST_BASE_NODE_WAIT =
      SystemProperties.getInt("com.datastax.driver.TEST_BASE_NODE_WAIT", 60);

  public static void setValue(SettableByIndexData<?> data, int i, DataType type, Object value) {
    switch (type.getName()) {
      case ASCII:
        data.setString(i, (String) value);
        break;
      case BIGINT:
        data.setLong(i, (Long) value);
        break;
      case BLOB:
        data.setBytes(i, (ByteBuffer) value);
        break;
      case BOOLEAN:
        data.setBool(i, (Boolean) value);
        break;
      case COUNTER:
        // Just a no-op, we shouldn't handle counters the same way than other types
        break;
      case DECIMAL:
        data.setDecimal(i, (BigDecimal) value);
        break;
      case DOUBLE:
        data.setDouble(i, (Double) value);
        break;
      case FLOAT:
        data.setFloat(i, (Float) value);
        break;
      case INET:
        data.setInet(i, (InetAddress) value);
        break;
      case TINYINT:
        data.setByte(i, (Byte) value);
        break;
      case SMALLINT:
        data.setShort(i, (Short) value);
        break;
      case INT:
        data.setInt(i, (Integer) value);
        break;
      case TEXT:
        data.setString(i, (String) value);
        break;
      case TIMESTAMP:
        data.setTimestamp(i, (Date) value);
        break;
      case DATE:
        data.setDate(i, (LocalDate) value);
        break;
      case TIME:
        data.setTime(i, (Long) value);
        break;
      case UUID:
        data.setUUID(i, (UUID) value);
        break;
      case VARCHAR:
        data.setString(i, (String) value);
        break;
      case VARINT:
        data.setVarint(i, (BigInteger) value);
        break;
      case TIMEUUID:
        data.setUUID(i, (UUID) value);
        break;
      case LIST:
        data.setList(i, (List<?>) value);
        break;
      case SET:
        data.setSet(i, (Set<?>) value);
        break;
      case MAP:
        data.setMap(i, (Map<?, ?>) value);
        break;
      default:
        throw new RuntimeException("Missing handling of " + type);
    }
  }

  @SuppressWarnings({"unchecked", "rawtypes"})
  public static void setValue(
      SettableByNameData<?> data, String name, DataType type, Object value) {
    switch (type.getName()) {
      case ASCII:
        data.setString(name, (String) value);
        break;
      case BIGINT:
        data.setLong(name, (Long) value);
        break;
      case BLOB:
        data.setBytes(name, (ByteBuffer) value);
        break;
      case BOOLEAN:
        data.setBool(name, (Boolean) value);
        break;
      case COUNTER:
        // Just a no-op, we shouldn't handle counters the same way than other types
        break;
      case DECIMAL:
        data.setDecimal(name, (BigDecimal) value);
        break;
      case DOUBLE:
        data.setDouble(name, (Double) value);
        break;
      case FLOAT:
        data.setFloat(name, (Float) value);
        break;
      case INET:
        data.setInet(name, (InetAddress) value);
        break;
      case TINYINT:
        data.setByte(name, (Byte) value);
        break;
      case SMALLINT:
        data.setShort(name, (Short) value);
        break;
      case INT:
        data.setInt(name, (Integer) value);
        break;
      case TEXT:
        data.setString(name, (String) value);
        break;
      case TIMESTAMP:
        data.setTimestamp(name, (Date) value);
        break;
      case DATE:
        data.setDate(name, (LocalDate) value);
        break;
      case TIME:
        data.setTime(name, (Long) value);
        break;
      case UUID:
        data.setUUID(name, (UUID) value);
        break;
      case VARCHAR:
        data.setString(name, (String) value);
        break;
      case VARINT:
        data.setVarint(name, (BigInteger) value);
        break;
      case TIMEUUID:
        data.setUUID(name, (UUID) value);
        break;
      case LIST:
        data.setList(name, (List) value);
        break;
      case SET:
        data.setSet(name, (Set) value);
        break;
      case MAP:
        data.setMap(name, (Map) value);
        break;
      default:
        throw new RuntimeException("Missing handling of " + type);
    }
  }

  public static Object getValue(
      GettableByIndexData data, int i, DataType type, CodecRegistry codecRegistry) {
    switch (type.getName()) {
      case ASCII:
        return data.getString(i);
      case BIGINT:
        return data.getLong(i);
      case BLOB:
        return data.getBytes(i);
      case BOOLEAN:
        return data.getBool(i);
      case COUNTER:
        return data.getLong(i);
      case DECIMAL:
        return data.getDecimal(i);
      case DOUBLE:
        return data.getDouble(i);
      case FLOAT:
        return data.getFloat(i);
      case INET:
        return data.getInet(i);
      case TINYINT:
        return data.getByte(i);
      case SMALLINT:
        return data.getShort(i);
      case INT:
        return data.getInt(i);
      case TEXT:
        return data.getString(i);
      case TIMESTAMP:
        return data.getTimestamp(i);
      case DATE:
        return data.getDate(i);
      case TIME:
        return data.getTime(i);
      case UUID:
        return data.getUUID(i);
      case VARCHAR:
        return data.getString(i);
      case VARINT:
        return data.getVarint(i);
      case TIMEUUID:
        return data.getUUID(i);
      case LIST:
        Class<?> listEltClass =
            codecRegistry.codecFor(type.getTypeArguments().get(0)).getJavaType().getRawType();
        return data.getList(i, listEltClass);
      case SET:
        Class<?> setEltClass =
            codecRegistry.codecFor(type.getTypeArguments().get(0)).getJavaType().getRawType();
        return data.getSet(i, setEltClass);
      case MAP:
        Class<?> keyClass =
            codecRegistry.codecFor(type.getTypeArguments().get(0)).getJavaType().getRawType();
        Class<?> valueClass =
            codecRegistry.codecFor(type.getTypeArguments().get(1)).getJavaType().getRawType();
        return data.getMap(i, keyClass, valueClass);
    }
    throw new RuntimeException("Missing handling of " + type);
  }

  public static Object getValue(
      GettableByNameData data, String name, DataType type, CodecRegistry codecRegistry) {
    switch (type.getName()) {
      case ASCII:
        return data.getString(name);
      case BIGINT:
        return data.getLong(name);
      case BLOB:
        return data.getBytes(name);
      case BOOLEAN:
        return data.getBool(name);
      case COUNTER:
        return data.getLong(name);
      case DECIMAL:
        return data.getDecimal(name);
      case DOUBLE:
        return data.getDouble(name);
      case FLOAT:
        return data.getFloat(name);
      case INET:
        return data.getInet(name);
      case TINYINT:
        return data.getByte(name);
      case SMALLINT:
        return data.getShort(name);
      case INT:
        return data.getInt(name);
      case TEXT:
        return data.getString(name);
      case TIMESTAMP:
        return data.getTimestamp(name);
      case DATE:
        return data.getDate(name);
      case TIME:
        return data.getTime(name);
      case UUID:
        return data.getUUID(name);
      case VARCHAR:
        return data.getString(name);
      case VARINT:
        return data.getVarint(name);
      case TIMEUUID:
        return data.getUUID(name);
      case LIST:
        Class<?> listEltClass =
            codecRegistry.codecFor(type.getTypeArguments().get(0)).getJavaType().getRawType();
        return data.getList(name, listEltClass);
      case SET:
        Class<?> setEltClass =
            codecRegistry.codecFor(type.getTypeArguments().get(0)).getJavaType().getRawType();
        return data.getSet(name, setEltClass);
      case MAP:
        Class<?> keyClass =
            codecRegistry.codecFor(type.getTypeArguments().get(0)).getJavaType().getRawType();
        Class<?> valueClass =
            codecRegistry.codecFor(type.getTypeArguments().get(1)).getJavaType().getRawType();
        return data.getMap(name, keyClass, valueClass);
    }
    throw new RuntimeException("Missing handling of " + type);
  }

  // Always return the "same" value for each type
  @SuppressWarnings("serial")
  public static Object getFixedValue(final DataType type) {
    try {
      switch (type.getName()) {
        case ASCII:
          return "An ascii string";
        case BIGINT:
          return 42L;
        case BLOB:
          return ByteBuffer.wrap(new byte[] {(byte) 4, (byte) 12, (byte) 1});
        case BOOLEAN:
          return true;
        case COUNTER:
          throw new UnsupportedOperationException("Cannot 'getSomeValue' for counters");
        case DURATION:
          return Duration.from("1h20m3s");
        case DECIMAL:
          return new BigDecimal(
              "3.1415926535897932384626433832795028841971693993751058209749445923078164062862089986280348253421170679");
        case DOUBLE:
          return 3.142519;
        case FLOAT:
          return 3.142519f;
        case INET:
          return InetAddress.getByAddress(new byte[] {(byte) 127, (byte) 0, (byte) 0, (byte) 1});
        case TINYINT:
          return (byte) 25;
        case SMALLINT:
          return (short) 26;
        case INT:
          return 24;
        case TEXT:
          return "A text string";
        case TIMESTAMP:
          return new Date(1352288289L);
        case DATE:
          return LocalDate.fromDaysSinceEpoch(0);
        case TIME:
          return 54012123450000L;
        case UUID:
          return UUID.fromString("087E9967-CCDC-4A9B-9036-05930140A41B");
        case VARCHAR:
          return "A varchar string";
        case VARINT:
          return new BigInteger("123456789012345678901234567890");
        case TIMEUUID:
          return UUID.fromString("FE2B4360-28C6-11E2-81C1-0800200C9A66");
        case LIST:
          return new ArrayList<Object>() {
            {
              add(getFixedValue(type.getTypeArguments().get(0)));
            }
          };
        case SET:
          return new HashSet<Object>() {
            {
              add(getFixedValue(type.getTypeArguments().get(0)));
            }
          };
        case MAP:
          return new HashMap<Object, Object>() {
            {
              put(
                  getFixedValue(type.getTypeArguments().get(0)),
                  getFixedValue(type.getTypeArguments().get(1)));
            }
          };
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    throw new RuntimeException("Missing handling of " + type);
  }

  // Always return the "same" value for each type
  @SuppressWarnings("serial")
  public static Object getFixedValue2(final DataType type) {
    try {
      switch (type.getName()) {
        case ASCII:
          return "A different ascii string";
        case BIGINT:
          return Long.MAX_VALUE;
        case BLOB:
          ByteBuffer bb = ByteBuffer.allocate(64);
          bb.putInt(0xCAFE);
          bb.putShort((short) 3);
          bb.putShort((short) 45);
          return bb;
        case BOOLEAN:
          return false;
        case COUNTER:
          throw new UnsupportedOperationException("Cannot 'getSomeValue' for counters");
        case DECIMAL:
          return new BigDecimal("12.3E+7");
        case DOUBLE:
          return Double.POSITIVE_INFINITY;
        case FLOAT:
          return Float.POSITIVE_INFINITY;
        case INET:
          return InetAddress.getByName("123.123.123.123");
        case TINYINT:
          return Byte.MAX_VALUE;
        case SMALLINT:
          return Short.MAX_VALUE;
        case INT:
          return Integer.MAX_VALUE;
        case TEXT:
          return "résumé";
        case TIMESTAMP:
          return new Date(872835240000L);
        case DATE:
          return LocalDate.fromDaysSinceEpoch(0);
        case TIME:
          return 54012123450000L;
        case UUID:
          return UUID.fromString("067e6162-3b6f-4ae2-a171-2470b63dff00");
        case VARCHAR:
          return "A different varchar résumé";
        case VARINT:
          return new BigInteger(Integer.toString(Integer.MAX_VALUE) + "000");
        case TIMEUUID:
          return UUID.fromString("FE2B4360-28C6-11E2-81C1-0800200C9A66");
        case LIST:
          return new ArrayList<Object>() {
            {
              add(getFixedValue2(type.getTypeArguments().get(0)));
            }
          };
        case SET:
          return new HashSet<Object>() {
            {
              add(getFixedValue2(type.getTypeArguments().get(0)));
            }
          };
        case MAP:
          return new HashMap<Object, Object>() {
            {
              put(
                  getFixedValue2(type.getTypeArguments().get(0)),
                  getFixedValue2(type.getTypeArguments().get(1)));
            }
          };
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    throw new RuntimeException("Missing handling of " + type);
  }

  /**
   * Returns a set of all primitive types supported by the given protocolVersion.
   *
   * <p>Primitive types are defined as the types that don't have type arguments (that is excluding
   * lists, sets, and maps, tuples and udts).
   *
   * @param protocolVersion protocol version to get types for.
   * @return returns a set of all the primitive types for the given protocolVersion.
   */
  static Set<DataType> allPrimitiveTypes(final ProtocolVersion protocolVersion) {
    return Sets.filter(
        DataType.allPrimitiveTypes(),
        new Predicate<DataType>() {

          @Override
          public boolean apply(DataType dataType) {
            return protocolVersion.compareTo(dataType.getName().minProtocolVersion) >= 0;
          }
        });
  }

  // Wait for a node to be up and running
  // This is used because there is some delay between when a node has been
  // added through ccm and when it's actually available for querying
  public static void waitForUp(String node, Cluster cluster) {
    waitFor(node, cluster, TEST_BASE_NODE_WAIT, false);
  }

  public static void waitForUp(String node, Cluster cluster, int timeoutSeconds) {
    waitFor(node, cluster, timeoutSeconds, false);
  }

  public static void waitForDown(String node, Cluster cluster) {
    waitFor(node, cluster, TEST_BASE_NODE_WAIT * 3, true);
  }

  public static void waitForDown(String node, Cluster cluster, int timeoutSeconds) {
    waitFor(node, cluster, timeoutSeconds, true);
  }

  private static void waitFor(
      String node, Cluster cluster, int timeoutSeconds, boolean waitForDown) {
    if (waitForDown) logger.debug("Waiting for node to leave: {}", node);
    else logger.debug("Waiting for upcoming node: {}", node);
    // In the case where the we've killed the last node in the cluster, if we haven't
    // tried doing an actual query, the driver won't realize that last node is dead until
    // keep alive kicks in, but that's a fairly long time. So we cheat and trigger a force
    // the detection by forcing a request.
    if (waitForDown)
      Futures.getUnchecked(cluster.manager.submitSchemaRefresh(null, null, null, null));
    if (waitForDown) {
      check()
          .every(1, SECONDS)
          .before(timeoutSeconds, SECONDS)
          .that(new HostIsDown(cluster, node))
          .becomesTrue();
    } else {
      check()
          .every(1, SECONDS)
          .before(timeoutSeconds, SECONDS)
          .that(new HostIsUp(cluster, node))
          .becomesTrue();
    }
  }

  private static class HostIsDown implements Callable<Boolean> {

    private final Cluster cluster;

    private final String ip;

    public HostIsDown(Cluster cluster, String ip) {
      this.cluster = cluster;
      this.ip = ip;
    }

    @Override
    public Boolean call() throws Exception {
      final Host host = findHost(cluster, ip);
      return host == null || !host.isUp();
    }
  }

  private static class HostIsUp implements Callable<Boolean> {

    private final Cluster cluster;

    private final String ip;

    public HostIsUp(Cluster cluster, String ip) {
      this.cluster = cluster;
      this.ip = ip;
    }

    @Override
    public Boolean call() throws Exception {
      final Host host = findHost(cluster, ip);
      return host != null && host.isUp();
    }
  }

  /**
   * Returns the IP of the {@code nth} host in the cluster (counting from 1, i.e., {@code
   * ipOfNode(1)} returns the IP of the first node.
   *
   * <p>In multi-DC setups, nodes are numbered in ascending order of their datacenter number. E.g.
   * with 2 DCs and 3 nodes in each DC, the first node in DC 2 is number 4.
   *
   * @return the IP of the {@code nth} host in the cluster.
   */
  public static String ipOfNode(int n) {
    return IP_PREFIX + n;
  }

  /**
   * Returns the address of the {@code nth} host in the cluster (counting from 1, i.e., {@code
   * ipOfNode(1)} returns the address of the first node.
   *
   * <p>In multi-DC setups, nodes are numbered in ascending order of their datacenter number. E.g.
   * with 2 DCs and 3 nodes in each DC, the first node in DC 2 is number 4.
   *
   * @return the IP of the {@code nth} host in the cluster.
   */
  public static InetAddress addressOfNode(int i) {
    try {
      return InetAddress.getByName(IP_PREFIX + i);
    } catch (UnknownHostException e) {
      // should never happen
      throw Throwables.propagate(e);
    }
  }

  public static Host findOrWaitForHost(Cluster cluster, int node, long duration, TimeUnit unit) {
    return findOrWaitForHost(cluster, ipOfNode(node), duration, unit);
  }

  public static Host findOrWaitForHost(
      final Cluster cluster, final String address, long duration, TimeUnit unit) {
    Host host = findHost(cluster, address);
    if (host == null) {
      final CountDownLatch addSignal = new CountDownLatch(1);
      Host.StateListener addListener =
          new StateListenerBase() {
            @Override
            public void onAdd(Host host) {
              if (host.getEndPoint().resolve().getAddress().getHostAddress().equals(address)) {
                // for a new node, because of this we also listen for add events.
                addSignal.countDown();
              }
            }
          };
      cluster.register(addListener);
      try {
        // Wait until an add event occurs or we timeout.
        if (addSignal.await(duration, unit)) {
          host = findHost(cluster, address);
        }
      } catch (InterruptedException e) {
        return null;
      } finally {
        cluster.unregister(addListener);
      }
    }
    return host;
  }

  public static Host findHost(Cluster cluster, int hostNumber) {
    return findHost(cluster, ipOfNode(hostNumber));
  }

  public static Host findHost(Cluster cluster, String address) {
    // Note: we can't rely on ProtocolOptions.getPort() to build an InetSocketAddress and call
    // metadata.getHost,
    // because the port doesn't have the correct value if addContactPointsWithPorts was used to
    // create the Cluster
    // (JAVA-860 will solve that)
    for (Host host : cluster.getMetadata().allHosts()) {
      if (host.getEndPoint().resolve().getAddress().getHostAddress().equals(address)) return host;
    }
    return null;
  }

  public static Host findOrWaitForControlConnection(
      final Cluster cluster, long duration, TimeUnit unit) {
    ControlConnection controlConnection = cluster.manager.controlConnection;
    long durationNs = TimeUnit.NANOSECONDS.convert(duration, unit);
    long start = System.nanoTime();
    while (System.nanoTime() - start < durationNs) {
      if (controlConnection.isOpen()) {
        return controlConnection.connectedHost();
      }
      Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
    }
    return null;
  }

  public static HostConnectionPool poolOf(Session session, int hostNumber) {
    SessionManager sessionManager = (SessionManager) session;
    return sessionManager.pools.get(findHost(session.getCluster(), hostNumber));
  }

  public static int numberOfLocalCoreConnections(Cluster cluster) {
    Configuration configuration = cluster.getConfiguration();
    return configuration.getPoolingOptions().getCoreConnectionsPerHost(HostDistance.LOCAL);
  }

  /**
   * @return A Scassandra instance with an arbitrarily chosen binary port from 8042-8142 and admin
   *     port from 8052-8152.
   */
  public static Scassandra createScassandraServer() {
    int binaryPort = findAvailablePort();
    int adminPort = findAvailablePort();
    return ScassandraFactory.createServer(ipOfNode(1), binaryPort, ipOfNode(1), adminPort);
  }

  private static final ConcurrentMap<String, AtomicInteger> IDENTIFIERS =
      new ConcurrentHashMap<String, AtomicInteger>();

  /**
   * Generates a unique CQL identifier with the given prefix.
   *
   * @param prefix The prefix for the identifier
   * @return a unique CQL identifier.
   */
  public static String generateIdentifier(String prefix) {
    AtomicInteger seq = new AtomicInteger(0);
    AtomicInteger previous = IDENTIFIERS.putIfAbsent(prefix, seq);
    if (previous != null) seq = previous;
    return prefix + seq.incrementAndGet();
  }

  /**
   * Finds an available port in the ephemeral range. This is loosely inspired by Apache MINA's
   * AvailablePortFinder.
   *
   * @return A local port that is currently unused.
   */
  public static synchronized int findAvailablePort() throws RuntimeException {
    ServerSocket ss = null;
    try {
      // let the system pick an ephemeral port
      ss = new ServerSocket(0);
      ss.setReuseAddress(true);
      return ss.getLocalPort();
    } catch (IOException e) {
      throw Throwables.propagate(e);
    } finally {
      if (ss != null) {
        try {
          ss.close();
        } catch (IOException e) {
          Throwables.propagate(e);
        }
      }
    }
  }

  private static final Predicate<InetSocketAddress> PORT_IS_UP =
      new Predicate<InetSocketAddress>() {

        @Override
        public boolean apply(InetSocketAddress address) {
          return pingPort(address.getAddress(), address.getPort());
        }
      };

  public static void waitUntilPortIsUp(InetSocketAddress address) {
    check().before(5, MINUTES).that(address, PORT_IS_UP).becomesTrue();
  }

  public static void waitUntilPortIsDown(InetSocketAddress address) {
    check().before(5, MINUTES).that(address, PORT_IS_UP).becomesFalse();
  }

  public static boolean pingPort(InetAddress address, int port) {
    logger.debug("Trying {}:{}...", address, port);
    boolean connectionSuccessful = false;
    Socket socket = null;
    try {
      socket = new Socket(address, port);
      connectionSuccessful = true;
      logger.debug("Successfully connected");
    } catch (IOException e) {
      logger.debug("Connection failed");
    } finally {
      if (socket != null)
        try {
          socket.close();
        } catch (IOException e) {
          logger.warn("Error closing socket to " + address, e);
        }
    }
    return connectionSuccessful;
  }

  /**
   * @return a {@link Cluster} instance that connects only to the control host of the given cluster.
   */
  public static Cluster buildControlCluster(Cluster cluster, CCMAccess ccm) {
    Host controlHost = cluster.manager.controlConnection.connectedHost();
    List<InetSocketAddress> singleAddress =
        Collections.singletonList(controlHost.getEndPoint().resolve());
    return configureClusterBuilder(Cluster.builder(), ccm)
        .withLoadBalancingPolicy(new WhiteListPolicy(new RoundRobinPolicy(), singleAddress))
        .build();
  }

  /**
   * Configures the builder with one contact point and port matching the given CCM cluster.
   * Therefore it's not required to call {@link Cluster.Builder#addContactPoints}, it will be done
   * automatically.
   *
   * @return The cluster builder (for method chaining).
   */
  public static Builder configureClusterBuilder(Builder builder, CCMAccess ccm) {
    // add only one contact point to force node1 to become the control host; some tests rely on
    // that.
    return configureClusterBuilder(builder, ccm, ccm.getContactPoints().get(0));
  }

  /**
   * Configures the builder with binary port matching the given CCM cluster and with the given
   * contact points. Therefore it's not required to call {@link Cluster.Builder#addContactPoints},
   * it will be done automatically.
   *
   * @return The cluster builder (for method chaining).
   */
  public static Builder configureClusterBuilder(
      Builder builder, CCMAccess ccm, InetAddress... contactPoints) {
    builder
        // use a different codec registry for each cluster instance
        .withCodecRegistry(new CodecRegistry())
        .addContactPoints(contactPoints)
        .withPort(ccm.getBinaryPort());
    return builder;
  }

  /** @return a {@link QueryOptions} that disables debouncing by setting intervals to 0ms. */
  public static QueryOptions nonDebouncingQueryOptions() {
    return new QueryOptions()
        .setRefreshNodeIntervalMillis(0)
        .setRefreshNodeListIntervalMillis(0)
        .setRefreshSchemaIntervalMillis(0);
  }

  /**
   * A custom {@link NettyOptions} that shuts down the {@link EventLoopGroup} after no quiet time.
   * This is useful for tests that consistently close clusters as otherwise there is a 2 second
   * delay (from JAVA-914).
   */
  public static NettyOptions nonQuietClusterCloseOptions =
      new NettyOptions() {
        @Override
        public void onClusterClose(EventLoopGroup eventLoopGroup) {
          eventLoopGroup.shutdownGracefully(0, 15, SECONDS).syncUninterruptibly();
        }
      };

  /**
   * Executes a task and catches any exception.
   *
   * @param task The task to execute.
   * @param logException Whether to log the exception, if any.
   */
  public static void executeNoFail(Runnable task, boolean logException) {
    try {
      task.run();
    } catch (Exception e) {
      if (logException) logger.error(e.getMessage(), e);
    }
  }

  /**
   * Executes a task and catches any exception.
   *
   * @param task The task to execute.
   * @param logException Whether to log the exception, if any.
   */
  public static void executeNoFail(Callable<?> task, boolean logException) {
    try {
      task.call();
    } catch (Exception e) {
      if (logException) logger.error(e.getMessage(), e);
    } catch (AssertionError e) {
      if (logException) logger.error(e.getMessage(), e);
    }
  }

  /**
   * Returns the system's free memory in megabytes.
   *
   * <p>This includes the free physical memory + the free swap memory.
   */
  public static long getFreeMemoryMB() {
    OperatingSystemMXBean bean =
        (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
    return (bean.getFreePhysicalMemorySize() + bean.getFreeSwapSpaceSize()) / 1024 / 1024;
  }

  /**
   * Helper for generating a DynamicCompositeType {@link ByteBuffer} from the given parameters.
   *
   * <p>Any of params given as an Integer will be considered with a field name of 'i', any as String
   * will be considered with a field name of 's'.
   *
   * @param params params to serialize.
   * @return bytes representing a DynamicCompositeType.
   */
  public static ByteBuffer serializeForDynamicCompositeType(Object... params) {
    List<ByteBuffer> l = new ArrayList<ByteBuffer>();
    int size = 0;
    for (Object p : params) {
      if (p instanceof Integer) {
        ByteBuffer elt = ByteBuffer.allocate(2 + 2 + 4 + 1);
        elt.putShort((short) (0x8000 | 'i'));
        elt.putShort((short) 4);
        elt.putInt((Integer) p);
        elt.put((byte) 0);
        elt.flip();
        size += elt.remaining();
        l.add(elt);
      } else if (p instanceof String) {
        ByteBuffer bytes = ByteBuffer.wrap(((String) p).getBytes());
        ByteBuffer elt = ByteBuffer.allocate(2 + 2 + bytes.remaining() + 1);
        elt.putShort((short) (0x8000 | 's'));
        elt.putShort((short) bytes.remaining());
        elt.put(bytes);
        elt.put((byte) 0);
        elt.flip();
        size += elt.remaining();
        l.add(elt);
      } else {
        throw new RuntimeException();
      }
    }
    ByteBuffer res = ByteBuffer.allocate(size);
    for (ByteBuffer bb : l) res.put(bb);
    res.flip();
    return res;
  }

  /**
   * Helper for generating a Composite {@link ByteBuffer} from the given parameters.
   *
   * <p>Expects Integer and String types for parameters.
   *
   * @param params params to serialize.
   * @return bytes representing a CompositeType
   */
  public static ByteBuffer serializeForCompositeType(Object... params) {

    List<ByteBuffer> l = new ArrayList<ByteBuffer>();
    int size = 0;
    for (Object p : params) {
      if (p instanceof Integer) {
        ByteBuffer elt = ByteBuffer.allocate(2 + 4 + 1);
        elt.putShort((short) 4);
        elt.putInt((Integer) p);
        elt.put((byte) 0);
        elt.flip();
        size += elt.remaining();
        l.add(elt);
      } else if (p instanceof String) {
        ByteBuffer bytes = ByteBuffer.wrap(((String) p).getBytes());
        ByteBuffer elt = ByteBuffer.allocate(2 + bytes.remaining() + 1);
        elt.putShort((short) bytes.remaining());
        elt.put(bytes);
        elt.put((byte) 0);
        elt.flip();
        size += elt.remaining();
        l.add(elt);
      } else {
        throw new RuntimeException();
      }
    }
    ByteBuffer res = ByteBuffer.allocate(size);
    for (ByteBuffer bb : l) res.put(bb);
    res.flip();
    return res;
  }

  public static Level setLogLevel(Class<?> logger, Level newLevel) {
    return setLogLevel(logger.getName(), newLevel);
  }

  public static Level setLogLevel(Logger logger, Level newLevel) {
    return setLogLevel(logger.getName(), newLevel);
  }

  public static Level setLogLevel(String logger, Level newLevel) {
    org.apache.log4j.Logger log4jLogger = org.apache.log4j.Logger.getLogger(logger);
    Level oldLevel = log4jLogger.getLevel();
    log4jLogger.setLevel(newLevel);
    return oldLevel;
  }

  /**
   * Throws a {@link SkipException} if the input {@link CCMAccess} does not support compact storage
   * (C* 4.0+ or DSE 6.0+).
   *
   * @param ccm cluster to check against
   */
  public static void compactStorageSupportCheck(CCMAccess ccm) {
    if (ccm.getCassandraVersion().nextStable().compareTo(VersionNumber.parse("4.0")) >= 0) {
      throw new SkipException(
          "Compact tables are not allowed in Cassandra starting with 4.0 version");
    }
    if (ccm.getDSEVersion() != null
        && ccm.getDSEVersion().compareTo(VersionNumber.parse("6.0")) >= 0) {
      throw new SkipException("Compact tables are not allowed in DSE starting with 6.0 version");
    }
  }
}
