@Test public void testScan() { int startScanKeyNumber = new Random().nextInt(recordsToInsert - recordsToScan); Vector<HashMap<String, ByteIterator>> scannedValues = new Vector<>(); assertEquals("Scan transaction FAILED.", Status.OK, riakClient.scan(bucket, keyPrefix + Integer.toString(startScanKeyNumber), recordsToScan, null, scannedValues)); for (int i = 0; i < recordsToScan; i++) { assertEquals("Scan test FAILED: the current scanned key is NOT MATCHING the expected one.", createExpectedHashMap(startScanKeyNumber + i).toString(), scannedValues.get(i).toString()); } }
@Override public Status scan(String table, String startkey, int recordcount, Set<String> fields, Vector<HashMap<String, ByteIterator>> result) { if (strongConsistency && !performStrongConsistentScans) { return Status.NOT_IMPLEMENTED; } IntIndexQuery iiq = new IntIndexQuery .Builder(new Namespace(bucketType2i, table), "key", getKeyAsLong(startkey), Long.MAX_VALUE) .withMaxResults(recordcount) .withPaginationSort(true) .build(); Location location; RiakFuture<IntIndexQuery.Response, IntIndexQuery> future = riakClient.executeAsync(iiq); try { IntIndexQuery.Response response = future.get(transactionTimeLimit, TimeUnit.SECONDS); List<IntIndexQuery.Response.Entry> entries = response.getEntries(); if (entries.size() == 0) { if (debug) { System.err.println("Unable to scan any record starting from key " + startkey + ", aborting transaction. " + "Reason: NOT FOUND"); } return Status.NOT_FOUND; } for (IntIndexQuery.Response.Entry entry : entries) { if (strongConsistency) { location = new Location(new Namespace(bucketType, table), entry.getRiakObjectLocation().getKeyAsString()); } else { location = entry.getRiakObjectLocation(); } FetchValue fv = new FetchValue.Builder(location) .withOption(FetchValue.Option.R, rvalue) .build(); FetchValue.Response keyResponse = fetch(fv); if (keyResponse.isNotFound()) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: NOT FOUND"); } return Status.NOT_FOUND; } HashMap<String, ByteIterator> partialResult = new HashMap<>(); createResultHashMap(fields, keyResponse, partialResult); result.add(partialResult); } } catch (TimeoutException e) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to scan all records starting from key " + startkey + ", aborting transaction. " + "Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; }
RiakKVClient extends DB { @Override public Status scan(String table, String startkey, int recordcount, Set<String> fields, Vector<HashMap<String, ByteIterator>> result) { if (strongConsistency && !performStrongConsistentScans) { return Status.NOT_IMPLEMENTED; } IntIndexQuery iiq = new IntIndexQuery .Builder(new Namespace(bucketType2i, table), "key", getKeyAsLong(startkey), Long.MAX_VALUE) .withMaxResults(recordcount) .withPaginationSort(true) .build(); Location location; RiakFuture<IntIndexQuery.Response, IntIndexQuery> future = riakClient.executeAsync(iiq); try { IntIndexQuery.Response response = future.get(transactionTimeLimit, TimeUnit.SECONDS); List<IntIndexQuery.Response.Entry> entries = response.getEntries(); if (entries.size() == 0) { if (debug) { System.err.println("Unable to scan any record starting from key " + startkey + ", aborting transaction. " + "Reason: NOT FOUND"); } return Status.NOT_FOUND; } for (IntIndexQuery.Response.Entry entry : entries) { if (strongConsistency) { location = new Location(new Namespace(bucketType, table), entry.getRiakObjectLocation().getKeyAsString()); } else { location = entry.getRiakObjectLocation(); } FetchValue fv = new FetchValue.Builder(location) .withOption(FetchValue.Option.R, rvalue) .build(); FetchValue.Response keyResponse = fetch(fv); if (keyResponse.isNotFound()) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: NOT FOUND"); } return Status.NOT_FOUND; } HashMap<String, ByteIterator> partialResult = new HashMap<>(); createResultHashMap(fields, keyResponse, partialResult); result.add(partialResult); } } catch (TimeoutException e) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to scan all records starting from key " + startkey + ", aborting transaction. " + "Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; } }
RiakKVClient extends DB { @Override public Status scan(String table, String startkey, int recordcount, Set<String> fields, Vector<HashMap<String, ByteIterator>> result) { if (strongConsistency && !performStrongConsistentScans) { return Status.NOT_IMPLEMENTED; } IntIndexQuery iiq = new IntIndexQuery .Builder(new Namespace(bucketType2i, table), "key", getKeyAsLong(startkey), Long.MAX_VALUE) .withMaxResults(recordcount) .withPaginationSort(true) .build(); Location location; RiakFuture<IntIndexQuery.Response, IntIndexQuery> future = riakClient.executeAsync(iiq); try { IntIndexQuery.Response response = future.get(transactionTimeLimit, TimeUnit.SECONDS); List<IntIndexQuery.Response.Entry> entries = response.getEntries(); if (entries.size() == 0) { if (debug) { System.err.println("Unable to scan any record starting from key " + startkey + ", aborting transaction. " + "Reason: NOT FOUND"); } return Status.NOT_FOUND; } for (IntIndexQuery.Response.Entry entry : entries) { if (strongConsistency) { location = new Location(new Namespace(bucketType, table), entry.getRiakObjectLocation().getKeyAsString()); } else { location = entry.getRiakObjectLocation(); } FetchValue fv = new FetchValue.Builder(location) .withOption(FetchValue.Option.R, rvalue) .build(); FetchValue.Response keyResponse = fetch(fv); if (keyResponse.isNotFound()) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: NOT FOUND"); } return Status.NOT_FOUND; } HashMap<String, ByteIterator> partialResult = new HashMap<>(); createResultHashMap(fields, keyResponse, partialResult); result.add(partialResult); } } catch (TimeoutException e) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to scan all records starting from key " + startkey + ", aborting transaction. " + "Reason: " + e.toString()); } return Status.ERROR; } return Status.OK; }  }
RiakKVClient extends DB { @Override public Status scan(String table, String startkey, int recordcount, Set<String> fields, Vector<HashMap<String, ByteIterator>> result) { if (strongConsistency && !performStrongConsistentScans) { return Status.NOT_IMPLEMENTED; } IntIndexQuery iiq = new IntIndexQuery .Builder(new Namespace(bucketType2i, table), "key", getKeyAsLong(startkey), Long.MAX_VALUE) .withMaxResults(recordcount) .withPaginationSort(true) .build(); Location location; RiakFuture<IntIndexQuery.Response, IntIndexQuery> future = riakClient.executeAsync(iiq); try { IntIndexQuery.Response response = future.get(transactionTimeLimit, TimeUnit.SECONDS); List<IntIndexQuery.Response.Entry> entries = response.getEntries(); if (entries.size() == 0) { if (debug) { System.err.println("Unable to scan any record starting from key " + startkey + ", aborting transaction. " + "Reason: NOT FOUND"); } return Status.NOT_FOUND; } for (IntIndexQuery.Response.Entry entry : entries) { if (strongConsistency) { location = new Location(new Namespace(bucketType, table), entry.getRiakObjectLocation().getKeyAsString()); } else { location = entry.getRiakObjectLocation(); } FetchValue fv = new FetchValue.Builder(location) .withOption(FetchValue.Option.R, rvalue) .build(); FetchValue.Response keyResponse = fetch(fv); if (keyResponse.isNotFound()) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: NOT FOUND"); } return Status.NOT_FOUND; } HashMap<String, ByteIterator> partialResult = new HashMap<>(); createResultHashMap(fields, keyResponse, partialResult); result.add(partialResult); } } catch (TimeoutException e) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to scan all records starting from key " + startkey + ", aborting transaction. " + "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 scan(String table, String startkey, int recordcount, Set<String> fields, Vector<HashMap<String, ByteIterator>> result) { if (strongConsistency && !performStrongConsistentScans) { return Status.NOT_IMPLEMENTED; } IntIndexQuery iiq = new IntIndexQuery .Builder(new Namespace(bucketType2i, table), "key", getKeyAsLong(startkey), Long.MAX_VALUE) .withMaxResults(recordcount) .withPaginationSort(true) .build(); Location location; RiakFuture<IntIndexQuery.Response, IntIndexQuery> future = riakClient.executeAsync(iiq); try { IntIndexQuery.Response response = future.get(transactionTimeLimit, TimeUnit.SECONDS); List<IntIndexQuery.Response.Entry> entries = response.getEntries(); if (entries.size() == 0) { if (debug) { System.err.println("Unable to scan any record starting from key " + startkey + ", aborting transaction. " + "Reason: NOT FOUND"); } return Status.NOT_FOUND; } for (IntIndexQuery.Response.Entry entry : entries) { if (strongConsistency) { location = new Location(new Namespace(bucketType, table), entry.getRiakObjectLocation().getKeyAsString()); } else { location = entry.getRiakObjectLocation(); } FetchValue fv = new FetchValue.Builder(location) .withOption(FetchValue.Option.R, rvalue) .build(); FetchValue.Response keyResponse = fetch(fv); if (keyResponse.isNotFound()) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: NOT FOUND"); } return Status.NOT_FOUND; } HashMap<String, ByteIterator> partialResult = new HashMap<>(); createResultHashMap(fields, keyResponse, partialResult); result.add(partialResult); } } catch (TimeoutException e) { if (debug) { System.err.println("Unable to scan all requested records starting from key " + startkey + ", aborting " + "transaction. Reason: TIME OUT"); } return TIME_OUT; } catch (Exception e) { if (debug) { System.err.println("Unable to scan all records starting from key " + startkey + ", aborting transaction. " + "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();  }