/*
 * Copyright 2013 Basho Technologies Inc
 *
 * Licensed 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.basho.riak.client.core.operations.itest;

import com.basho.riak.client.core.operations.DtFetchOperation;
import com.basho.riak.client.core.operations.DtUpdateOperation;
import com.basho.riak.client.operations.crdt.MapMutation;
import com.basho.riak.client.query.crdt.ops.MapOp;
import com.basho.riak.client.query.crdt.types.*;
import com.basho.riak.client.util.ByteArrayWrapper;
import org.junit.Test;

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

import static com.basho.riak.client.operations.crdt.CounterMutation.increment;
import static com.basho.riak.client.operations.crdt.CrdtMutation.forMap;
import static com.basho.riak.client.operations.crdt.FlagMutation.enabled;
import static com.basho.riak.client.operations.crdt.RegisterMutation.registerValue;
import static junit.framework.Assert.*;

public class ITestCrdtApi extends ITestBase
{

    @Test
    public void simpleTest() throws ExecutionException, InterruptedException
    {

        /**
         * Update some info about a user in a table of users
         */

        resetAndEmptyBucket(bucketName, mapBucketType);

        // ByteArrayWrappers make it look messy, so define them all here.
        ByteArrayWrapper numLogins = ByteArrayWrapper.create("logins");
        ByteArrayWrapper lastLoginTime = ByteArrayWrapper.create("last-login");
        ByteBuffer nowBinary = ByteBuffer.allocate(8).putLong(System.currentTimeMillis());
        ByteArrayWrapper now = ByteArrayWrapper.create(nowBinary.array());
        ByteArrayWrapper username = ByteArrayWrapper.create("username");
        ByteArrayWrapper loggedIn = ByteArrayWrapper.create("logged-in");

        // Create a builder for the user and update the values
        MapMutation userMapUpdate = forMap()
            .update(numLogins, increment())
            .update(lastLoginTime, registerValue(now))
            .update(loggedIn, enabled());

        // Now create an update for the user's entry
        MapMutation userEntryUpdate = forMap()
            .update(username, userMapUpdate);

        ByteArrayWrapper key = ByteArrayWrapper.create("user-info");

        DtUpdateOperation update =
            new DtUpdateOperation(bucketName, mapBucketType)
                .withOp(userEntryUpdate.getOp())
                .withKey(key);

        cluster.execute(update);
        update.get();

        DtFetchOperation fetch = new DtFetchOperation(bucketName, key).withBucketType(mapBucketType);
        cluster.execute(fetch);

        // enclosing map
        CrdtElement element = fetch.get();
        assertNotNull(element);
        assertTrue(element.isMap());
        CrdtMap usersMap = element.getAsMap();

        // username
        CrdtElement usernameElement = usersMap.get(username);
        assertNotNull(usernameElement);
        assertTrue(usernameElement.isMap());
        CrdtMap usernameMap = usernameElement.getAsMap();

        // logins
        CrdtElement numLoginsElement = usernameMap.get(numLogins);
        assertNotNull(numLoginsElement);
        assertTrue(numLoginsElement.isCounter());
        CrdtCounter numLoginsCounter = numLoginsElement.getAsCounter();
        assertEquals(1, numLoginsCounter.getValue());

        // last-login
        CrdtElement lastLoginTimeElement = usernameMap.get(lastLoginTime);
        assertNotNull(lastLoginTimeElement);
        assertTrue(lastLoginTimeElement.isRegister());
        CrdtRegister lastLoginTimeRegister = lastLoginTimeElement.getAsRegister();
        assertEquals(now, lastLoginTimeRegister.getValue());

        // logged-in
        CrdtElement loggedInElement = usernameMap.get(loggedIn);
        assertNotNull(loggedInElement);
        assertTrue(loggedInElement.isFlag());
        CrdtFlag loggedInFlag = loggedInElement.getAsFlag();
        assertEquals(true, loggedInFlag.getEnabled());

    }

}

@@ -835,24 +835,82 @@
 apMutation;%0A
+import com.basho.riak.client.operations.crdt.SetMutation;%0A
 import com.b
@@ -1302,16 +1302,89 @@
 forMap;%0A
+import static com.basho.riak.client.operations.crdt.CrdtMutation.forSet;%0A
 import s
@@ -2401,64 +2401,715 @@
 %22);%0A
-%0A        // Create a builder for the user and update the
+        ByteArrayWrapper favoriteThings = ByteArrayWrapper.create(%22favs%22);%0A%0A%0A        /*%0A%0A            Data structure:%0A%0A            bucket-type == map%0A            -%3E %22username%22 : map%0A              -%3E %22logins%22     : counter%0A              -%3E %22last-login%22 : register%0A              -%3E %22logged-in%22  : flag%0A              -%3E %22favs%22       : set%0A%0A         */%0A%0A        // Build a set of things%0A        ByteBuffer buffer = ByteBuffer.allocate(4);%0A        SetMutation favorites = forSet();%0A        for (int i = 0; i %3C 10; ++i)%0A        %7B%0A            buffer.putInt(i);%0A            favorites.add(ByteArrayWrapper.create(buffer.array()));%0A            buffer.rewind();%0A        %7D%0A%0A        // Create an update for the user their
  val
@@ -3200,16 +3200,42 @@
 ement())
+                // counter
 %0A       
@@ -3281,16 +3281,32 @@
 ue(now))
+     // register
 %0A       
@@ -3334,25 +3334,116 @@
 , enabled())
-;
+                   // flag%0A            .update(favoriteThings, favorites);            // set
 %0A%0A        //
@@ -4032,21 +4032,13 @@
  // 
-enclosing map
+users
 %0A   
@@ -4437,16 +4437,26 @@
 / logins
+ - counter
 %0A       
@@ -4754,16 +4754,27 @@
 st-login
+ - register
 %0A       
@@ -5106,16 +5106,23 @@
 ogged-in
+ - flag
 %0A       

add set to example, in addition to some documentation