package mtdb;

import java.nio.ByteBuffer;
import java.util.concurrent.ThreadLocalRandom;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;

public class CassCli extends DbCli
{
	public static DbCli GetInstance() {
		if (_instance != null)
			return _instance;

		synchronized (_instance_mutex) {
			if (_instance == null) {
				_instance = new CassCli();
				return _instance;
			} else {
				return _instance;
			}
		}
	}

	private Cluster _cluster = null;
	private Session _session = null;

	@Override
	protected void DbInit() {
		try (Cons.MeasureTime _ = new Cons.MeasureTime("Connecting to Cassandra server ...")) {
			_cluster = Cluster.builder().addContactPoint(Conf.mutantsLoadgenOptions.global.server_addr).build();
			_session = _cluster.connect("mtdb1");
		}
	}

	@Override
	protected void DbClose() {
		if (_session != null)
			_session.close();

		// Cluster.close() takes 2.2 secs. System.exit() instead to save time. Not sure
		// what's the implication of this.
		//if (_cluster != null)
		//	_cluster.close();
	}

	@Override
	protected void DbWrite(Op op) {
		_session.execute(String.format(
					"INSERT INTO table1 (key, epoch_sec, c0) "
					+ "VALUES (%d, %d, 0x%s)"
					, op.wrs.key
					, op.wrs.wEpochSec
					, BuildC0(Conf.mutantsLoadgenOptions.per_obj.obj_size)
					));
	}

	@Override
	protected void DbRead(Op op) throws InterruptedException {
		ResultSet rs = _session.execute("SELECT * FROM table1 WHERE key=" + op.wrs.key);
		//Cons.P(String.format("fetched %d rows", rs.all().size()));
		int rsSize = rs.all().size();
		if (rsSize != 1) {
			// 0 is dangerous. It scans all SStables and some of them will cause the
			// bloom filter false positives.
			throw new RuntimeException("Unexpected rsSize=" + rsSize);
		}
	}

	// Generate random uncompressible data
	private String BuildC0(int length) {
		ThreadLocalRandom tlr = ThreadLocalRandom.current();
		// 64-bit, 8-byte
		long l = tlr.nextLong();

		ByteBuffer bb = ByteBuffer.allocate(length);
		//b.order(ByteOrder.BIG_ENDIAN); // optional, the initial order of a byte buffer is always BIG_ENDIAN.
		// bb is padded with 0s for the remainders of 8-byte blocks.
		for (int i = 0; i < (length / 8); i ++)
			bb.putLong(l);
		//Cons.P(Util.toHex(bb.array()));
		return Util.toHex(bb.array());
	}
}

@@ -121,24 +121,70 @@
 re.Cluster;%0A
+import com.datastax.driver.core.exceptions.*;%0A
 import com.d
@@ -1228,16 +1228,44 @@
 p op) %7B%0A
+%09%09while (true) %7B%0A%09%09%09try %7B%0A%09%09
 %09%09_sessi
@@ -1290,16 +1290,18 @@
 format(%0A
+%09%09
 %09%09%09%09%09%22IN
@@ -1345,16 +1345,18 @@
  %22%0A%09%09%09%09%09
+%09
+%09
 + %22VALUE
@@ -1374,24 +1374,26 @@
 0x%25s)%22%0A%09%09%09%09%09
+%09%09
 , op.wrs.key
@@ -1394,24 +1394,26 @@
 rs.key%0A%09%09%09%09%09
+%09
+%09
 , op.wrs.wEp
@@ -1419,16 +1419,18 @@
 pochSec%0A
+%09%09
 %09%09%09%09%09, B
@@ -1486,20 +1486,119 @@
 e)%0A%09%09%09%09%09
+%09%09
 ));%0A
+%09%09%09%09return;%0A%09%09%09%7D catch (WriteTimeoutException e) %7B%0A%09%09%09%09// Timeout. Retry until succeed.%0A%09%09%09%7D%0A%09%09%7D%0A
 %09%7D%0A%0A%09@Ov
@@ -1664,16 +1664,44 @@
 ption %7B%0A
+%09%09while (true) %7B%0A%09%09%09try %7B%0A%09%09
 %09%09Result
@@ -1775,16 +1775,18 @@
 s.key);%0A
+%09%09
 %09%09//Cons
@@ -1840,16 +1840,18 @@
 ze()));%0A
+%09%09
 %09%09int rs
@@ -1874,16 +1874,18 @@
 size();%0A
+%09%09
 %09%09if (rs
@@ -1897,16 +1897,18 @@
 != 1) %7B%0A
+%09%09
 %09%09%09// 0 
@@ -1978,16 +1978,18 @@
  the%0A%09%09%09
+%09%09
 // bloom
@@ -2013,16 +2013,18 @@
 itives.%0A
+%09%09
 %09%09%09throw
@@ -2077,16 +2077,114 @@
 sSize);%0A
+%09%09%09%09%7D%0A%09%09%09%09return;%0A%09%09%09%7D catch (ReadTimeoutException e) %7B%0A%09%09%09%09// Timeout. Retry until succeed.%0A%09%09%09%7D%0A
 %09%09%7D%0A%09%7D%0A%0A

handle timeout exceptions and retry