/*
 * This file is provided 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.basho.riak.client.itest;

import static org.junit.Assert.*;

import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Before;
import org.junit.Test;

import com.basho.riak.client.RiakClient;
import com.basho.riak.client.RiakObject;

public class ITestDataLoad {

    final String RIAK_URL = "http://127.0.0.1:8098/riak";
    final String BUCKET = "test_data_load";
    final int NUM_VALUES = 10;
    final int VALUE_LENGTH = 512;
    
    byte data[][] = new byte[NUM_VALUES][VALUE_LENGTH]; 
    
    @Before public void setup() {
        for (int i = 0; i < 10; i++) {
            new Random().nextBytes(data[i]);
        }
    }
    
    @Test public void concurrent_data_load() throws InterruptedException {
        
        final int NUM_THREADS = 5;
        final int NUM_OBJECTS = 200;
        final Thread[] threads = new Thread[NUM_THREADS];
        final AtomicInteger idx = new AtomicInteger(0);
        
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new Runnable() {
                
                public void run() {
                    RiakClient riak = new RiakClient(RIAK_URL);
                    Random rnd = new Random();
                    for (int i = 0; i < NUM_OBJECTS / NUM_THREADS; i++) {
                        String key = "data-load-" + idx.getAndIncrement();
                        String value = new String(data[rnd.nextInt(NUM_VALUES)]);
                        RiakObject o = riak.fetch(BUCKET, key).getObject();
                        if (o == null) {
                            o = new RiakObject(riak, BUCKET, key, value.getBytes());
                        } else {
                            o.setValue(value);
                        }
                        Utils.assertSuccess(o.store());
                    }
                }
            });
            threads[i].start();
        }
        
        for (Thread thread : threads) {
            thread.join();
        }
    }

    @Test public void vclock_doesnt_explode_using_single_client() {
        final RiakClient riak = new RiakClient(RIAK_URL);
        final RiakObject o = new RiakObject(riak, BUCKET, "test-vclock-size");
        final Random rnd = new Random();
        
        Utils.assertSuccess(o.delete(Utils.WRITE_3_REPLICAS()));
        Utils.assertSuccess(o.store());
        
        String originalVclock = o.getVclock();
        
        for (int i = 0; i < 15; i++) {
            o.fetch();
            o.setValue(data[rnd.nextInt(NUM_VALUES)]);
            Utils.assertSuccess(o.store());
        }

        assertEquals(originalVclock.length(), o.getVclock().length());
    }
}
@@ -671,16 +671,39 @@
 Random;%0A
+import java.util.UUID;%0A
 import j
@@ -2757,51 +2757,80 @@
 nal 
-RiakObject o = new RiakObject(riak, BUCKET,
+String bucket = UUID.randomUUID().toString();%0A        final String key =
  %22te
@@ -2844,17 +2844,16 @@
 ck-size%22
-)
 ;%0A      
@@ -2865,108 +2865,99 @@
 al R
-andom rnd = new Random();%0A        %0A        Utils.assertSuccess(o.delete(Utils.WRITE_3_REPLICAS())
+iakObject o = new RiakObject(riak, bucket, key);%0A        final Random rnd = new Random(
 );%0A
+%0A
     
@@ -2992,24 +2992,16 @@
 ore());%0A
-        
 %0A       
@@ -3040,24 +3040,16 @@
 lock();%0A
-        
 %0A       
@@ -3280,19 +3280,63 @@
 .length());%0A
+        o.delete(Utils.WRITE_3_REPLICAS());%0A
     %7D%0A%7D

Fix sporadically failing vclock test