@Test public void testInsert() { HashMap<String, String> insertValue = createExpectedHashMap(recordsToInsert); HashMap<String, ByteIterator> readValue = new HashMap<>(); assertEquals("Insert transaction FAILED.", Status.OK, riakClient.insert(bucket, keyPrefix + Integer.toString(recordsToInsert), StringByteIterator. getByteIteratorMap(insertValue))); assertEquals("Insert test FAILED. Unable to read inserted value.", Status.OK, riakClient.read(bucket, keyPrefix + Integer.toString(recordsToInsert), null, readValue)); assertEquals("Insert test FAILED. Actual read transaction value is NOT MATCHING the inserted one.", insertValue.toString(), readValue.toString()); }
@Override public Status insert(String table, String key, Map<String, ByteIterator> values) { Location location = new Location(new Namespace(bucketType, table), key); RiakObject object = new RiakObject(); if (strongConsistency && performStrongConsistentScans) { Location fakeLocation = new Location(new Namespace(strongConsistentScansBucketType, table), key); RiakObject fakeObject = new RiakObject(); fakeObject.setValue(BinaryValue.create(new byte[]{0x00})); fakeObject.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); StoreValue fakeStore = new StoreValue.Builder(fakeObject) .withLocation(fakeLocation) .build(); riakClient.executeAsync(fakeStore); } else if (!strongConsistency) { object.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); } object.setValue(BinaryValue.create(serializeTable(values))); StoreValue store = new StoreValue.Builder(object) .withOption(StoreValue.Option.W, wvalue) .withLocation(location) .build(); RiakFuture<StoreValue.Response, Location> future = riakClient.executeAsync(store); try { future.get(transactionTimeLimit, TimeUnit.SECONDS); } catch (TimeoutException e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; }
RiakKVClient extends DB { @Override public Status insert(String table, String key, Map<String, ByteIterator> values) { Location location = new Location(new Namespace(bucketType, table), key); RiakObject object = new RiakObject(); if (strongConsistency && performStrongConsistentScans) { Location fakeLocation = new Location(new Namespace(strongConsistentScansBucketType, table), key); RiakObject fakeObject = new RiakObject(); fakeObject.setValue(BinaryValue.create(new byte[]{0x00})); fakeObject.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); StoreValue fakeStore = new StoreValue.Builder(fakeObject) .withLocation(fakeLocation) .build(); riakClient.executeAsync(fakeStore); } else if (!strongConsistency) { object.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); } object.setValue(BinaryValue.create(serializeTable(values))); StoreValue store = new StoreValue.Builder(object) .withOption(StoreValue.Option.W, wvalue) .withLocation(location) .build(); RiakFuture<StoreValue.Response, Location> future = riakClient.executeAsync(store); try { future.get(transactionTimeLimit, TimeUnit.SECONDS); } catch (TimeoutException e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; } }
RiakKVClient extends DB { @Override public Status insert(String table, String key, Map<String, ByteIterator> values) { Location location = new Location(new Namespace(bucketType, table), key); RiakObject object = new RiakObject(); if (strongConsistency && performStrongConsistentScans) { Location fakeLocation = new Location(new Namespace(strongConsistentScansBucketType, table), key); RiakObject fakeObject = new RiakObject(); fakeObject.setValue(BinaryValue.create(new byte[]{0x00})); fakeObject.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); StoreValue fakeStore = new StoreValue.Builder(fakeObject) .withLocation(fakeLocation) .build(); riakClient.executeAsync(fakeStore); } else if (!strongConsistency) { object.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); } object.setValue(BinaryValue.create(serializeTable(values))); StoreValue store = new StoreValue.Builder(object) .withOption(StoreValue.Option.W, wvalue) .withLocation(location) .build(); RiakFuture<StoreValue.Response, Location> future = riakClient.executeAsync(store); try { future.get(transactionTimeLimit, TimeUnit.SECONDS); } catch (TimeoutException e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; }  }
RiakKVClient extends DB { @Override public Status insert(String table, String key, Map<String, ByteIterator> values) { Location location = new Location(new Namespace(bucketType, table), key); RiakObject object = new RiakObject(); if (strongConsistency && performStrongConsistentScans) { Location fakeLocation = new Location(new Namespace(strongConsistentScansBucketType, table), key); RiakObject fakeObject = new RiakObject(); fakeObject.setValue(BinaryValue.create(new byte[]{0x00})); fakeObject.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); StoreValue fakeStore = new StoreValue.Builder(fakeObject) .withLocation(fakeLocation) .build(); riakClient.executeAsync(fakeStore); } else if (!strongConsistency) { object.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); } object.setValue(BinaryValue.create(serializeTable(values))); StoreValue store = new StoreValue.Builder(object) .withOption(StoreValue.Option.W, wvalue) .withLocation(location) .build(); RiakFuture<StoreValue.Response, Location> future = riakClient.executeAsync(store); try { future.get(transactionTimeLimit, TimeUnit.SECONDS); } catch (TimeoutException e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; }  void init(); @Override Status read(String table, String key, Set<String> fields, Map<String, ByteIterator> result); @Override Status scan(String table, String startkey, int recordcount, Set<String> fields,
                     Vector<HashMap<String, ByteIterator>> result); @Override Status insert(String table, String key, Map<String, ByteIterator> values); @Override Status update(String table, String key, Map<String, ByteIterator> values); @Override Status delete(String table, String key); void cleanup(); }
RiakKVClient extends DB { @Override public Status insert(String table, String key, Map<String, ByteIterator> values) { Location location = new Location(new Namespace(bucketType, table), key); RiakObject object = new RiakObject(); if (strongConsistency && performStrongConsistentScans) { Location fakeLocation = new Location(new Namespace(strongConsistentScansBucketType, table), key); RiakObject fakeObject = new RiakObject(); fakeObject.setValue(BinaryValue.create(new byte[]{0x00})); fakeObject.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); StoreValue fakeStore = new StoreValue.Builder(fakeObject) .withLocation(fakeLocation) .build(); riakClient.executeAsync(fakeStore); } else if (!strongConsistency) { object.getIndexes().getIndex(LongIntIndex.named("key_int")).add(getKeyAsLong(key)); } object.setValue(BinaryValue.create(serializeTable(values))); StoreValue store = new StoreValue.Builder(object) .withOption(StoreValue.Option.W, wvalue) .withLocation(location) .build(); RiakFuture<StoreValue.Response, Location> future = riakClient.executeAsync(store); try { future.get(transactionTimeLimit, TimeUnit.SECONDS); } catch (TimeoutException e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to insert key " + key + ". Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; }  void init(); @Override Status read(String table, String key, Set<String> fields, Map<String, ByteIterator> result); @Override Status scan(String table, String startkey, int recordcount, Set<String> fields,
                     Vector<HashMap<String, ByteIterator>> result); @Override Status insert(String table, String key, Map<String, ByteIterator> values); @Override Status update(String table, String key, Map<String, ByteIterator> values); @Override Status delete(String table, String key); void cleanup();  }