package com.basho.riak.client.core.netty;

import com.basho.riak.client.core.FutureOperation;
import com.basho.riak.client.core.RiakNode;
import com.basho.riak.client.core.operations.ts.FetchOperation;
import com.basho.riak.client.core.operations.ts.QueryOperation;
import com.basho.riak.client.core.operations.ts.StoreOperation;
import com.basho.riak.client.core.query.timeseries.Cell;
import com.basho.riak.client.core.query.timeseries.QueryResult;
import com.basho.riak.client.core.query.timeseries.Row;
import com.basho.riak.client.core.util.BinaryValue;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;

import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;

/**
 * @author Sergey Galkin <srggal at gmail dot com>
 */
public class ITestTTBNode {

    protected final static String tableName = "GeoCheckin";

    protected final static long now = 1443806900000l; // "now"
    protected final static long fiveMinsInMS = 5l * 60l * 1000l;
    protected final static long fiveMinsAgo = now - fiveMinsInMS;
    protected final static long tenMinsAgo = fiveMinsAgo - fiveMinsInMS;
    protected final static long fifteenMinsAgo = tenMinsAgo - fiveMinsInMS;
    protected final static long fifteenMinsInFuture = now + (fiveMinsInMS * 3l);

    private RiakNode riakNode;
    protected final static List<Row> rows = Arrays.asList(
            // "Normal" Data
            new Row(new Cell("hash1"), new Cell("user1"), Cell.newTimestamp(fifteenMinsAgo), new Cell("cloudy"), new Cell(79.0), new Cell(1), new Cell(true)),
            new Row(new Cell("hash1"), new Cell("user1"), Cell.newTimestamp(fiveMinsAgo), new Cell("sunny"),  new Cell(80.5), new Cell(2), new Cell(true)),
            new Row(new Cell("hash1"), new Cell("user1"), Cell.newTimestamp(now), new Cell("sunny"),  new Cell(81.0), new Cell(10), new Cell(false)),

            // Null Cell row
            new Row(new Cell("hash1"), new Cell("user2"), Cell.newTimestamp(fiveMinsAgo), new Cell("cloudy"), null, null, new Cell(true)),

            // Data for single reads / deletes
            new Row(new Cell("hash2"), new Cell("user4"), Cell.newTimestamp(fifteenMinsAgo), new Cell("rain"), new Cell(79.0), new Cell(2), new Cell(false)),
            new Row(new Cell("hash2"), new Cell("user4"), Cell.newTimestamp(fiveMinsAgo), new Cell("wind"),  new Cell(50.5), new Cell(3), new Cell(true)),
            new Row(new Cell("hash2"), new Cell("user4"), Cell.newTimestamp(now), new Cell("snow"),  new Cell(20.0), new Cell(11), new Cell(true)));


    @Before
    public void setup() throws UnknownHostException {
        riakNode = new RiakNode.Builder()
                .useTTB()
                .withRemotePort(10017)
                .build();

        riakNode.start();
    }


    @After
    public void shutdonw(){
        if (riakNode != null) {
            riakNode.shutdown();
        }
    }

    @Test
    public void storeData() throws InterruptedException, ExecutionException {
        storeDataInternal(tableName, rows.subList(0,1));
    }

    @Test
    public void getData() throws ExecutionException, InterruptedException {
        storeDataInternal(tableName, rows);

        for (Row r: rows){
            final FetchOperation fetch = new FetchOperation.Builder(BinaryValue.create(tableName),
                    r.getCells().subList(0,3) // use the only PK values
                ).build();

            QueryResult queryResult = execute(fetch);

            assertEquals(1, queryResult.getRows().size());
            //assertArrayEquals(r.getCells().toArray(), queryResult.getRows().get(0).getCells().toArray());
        }
    }

    @Ignore
    @Test
    public void queryData() throws InterruptedException, ExecutionException {
        storeDataInternal(tableName, rows);

        final String queryText = "select * from GeoCheckin " +
                "where user = 'user1' and " +
                "geohash = 'hash1' and " +
                "(time = " + tenMinsAgo +" and " +
                "(time < "+ now + ")) ";

        final QueryOperation query = new QueryOperation.Builder(BinaryValue.create(queryText)).build();
        final QueryResult queryResult = execute(query);

        assertEquals(7, queryResult.getColumnDescriptions().size());
        assertEquals(1, queryResult.getRows().size());
    }

    @Test
    public void storeDatatoNoneExistentTable() throws InterruptedException, ExecutionException {
        try {
            storeDataInternal(UUID.randomUUID().toString(), rows);
        } catch (ExecutionException e) {
            if (e.getCause() instanceof RiakResponseException
                    && e.getCause().getMessage().startsWith("Bucket type")
                    && e.getCause().getMessage().endsWith("is missing.")) {
                // It's OK
                return;
            }
            throw e;
        }
    }

    private <T, U, S, O extends FutureOperation<T,U,S>> T execute(O operation) throws ExecutionException, InterruptedException {
        riakNode.execute((FutureOperation) operation);
        return (T)((FutureOperation) operation).get();
    }

    private void storeDataInternal(String table, List<Row> rows) throws ExecutionException, InterruptedException {
        final StoreOperation store = new StoreOperation.Builder(BinaryValue.create(table)).withRows(rows).build();

        execute(store);
    }
}

@@ -3476,35 +3476,16 @@
 Builder(
-BinaryValue.create(
 tableNam
@@ -3485,17 +3485,16 @@
 ableName
-)
 ,%0A      
@@ -3517,16 +3517,20 @@
 getCells
+Copy
 ().subLi
@@ -3685,39 +3685,37 @@
 ryResult.getRows
-().size
+Count
 ());%0A           
@@ -4282,35 +4282,16 @@
 Builder(
-BinaryValue.create(
 queryTex
@@ -4292,17 +4292,16 @@
 eryText)
-)
 .build()
@@ -4416,16 +4416,20 @@
 riptions
+Copy
 ().size(
@@ -4475,23 +4475,21 @@
 .getRows
-().size
+Count
 ());%0A   
@@ -4848,18 +4848,24 @@
 th(%22
-Bucket typ
+Time Series tabl
 e%22)%0A
@@ -4927,18 +4927,22 @@
 th(%22
-is missing
+does not exist
 .%22))
@@ -5476,34 +5476,14 @@
 der(
-BinaryValue.create(
 table)
-)
 .wit

Fix broken tests