Java 类com.amazonaws.services.dynamodbv2.model.ScanResult 实例源码

/**
 * Gets all.
 *
 * @return the all
 */
public Collection<Ticket> getAll() {
    final Collection<Ticket> tickets = new ArrayList<>();
    final Collection<TicketDefinition> metadata = this.ticketCatalog.findAll();
    metadata.forEach(r -> {
        final ScanRequest scan = new ScanRequest(r.getProperties().getStorageName());
        LOGGER.debug("Scanning table with request [{}]", scan);
        final ScanResult result = this.amazonDynamoDBClient.scan(scan);
        LOGGER.debug("Scanned table with result [{}]", scan);

        tickets.addAll(result.getItems()
                .stream()
                .map(DynamoDbTicketRegistryFacilitator::deserializeTicket)
                .collect(Collectors.toList()));
    });
    return tickets;
}

@Override
public PropertySource<?> locate(final Environment environment) {
    final AmazonDynamoDBClient amazonDynamoDBClient = getAmazonDynamoDbClient(environment);
    createSettingsTable(amazonDynamoDBClient, false);

    final ScanRequest scan = new ScanRequest(TABLE_NAME);
    LOGGER.debug("Scanning table with request [{}]", scan);
    final ScanResult result = amazonDynamoDBClient.scan(scan);
    LOGGER.debug("Scanned table with result [{}]", scan);

    final Properties props = new Properties();
    result.getItems()
            .stream()
            .map(DynamoDbCloudConfigBootstrapConfiguration::retrieveSetting)
            .forEach(p -> props.put(p.getKey(), p.getValue()));
    return new PropertiesPropertySource(getClass().getSimpleName(), props);
}

public RetryResult<ScanResult> scanTable(
    String tableName, DynamoDBQueryFilter dynamoDBQueryFilter, Integer segment, Integer
    totalSegments, Map<String, AttributeValue> exclusiveStartKey, long limit, Reporter reporter) {
  final ScanRequest scanRequest = new ScanRequest(tableName)
      .withExclusiveStartKey(exclusiveStartKey)
      .withLimit(Ints.checkedCast(limit))
      .withSegment(segment)
      .withTotalSegments(totalSegments)
      .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);

  if (dynamoDBQueryFilter != null) {
    Map<String, Condition> scanFilter = dynamoDBQueryFilter.getScanFilter();
    if (!scanFilter.isEmpty()) {
      scanRequest.setScanFilter(scanFilter);
    }
  }

  RetryResult<ScanResult> retryResult = getRetryDriver().runWithRetry(new Callable<ScanResult>() {
    @Override
    public ScanResult call() {
      log.debug("Executing DynamoDB scan: " + scanRequest);
      return dynamoDB.scan(scanRequest);
    }
  }, reporter, PrintCounter.DynamoDBReadThrottle);
  return retryResult;
}

@Override
protected PageResults<Map<String, AttributeValue>> fetchPage(RequestLimit lim) {
  // Read from DynamoDB
  RetryResult<ScanResult> retryResult = context.getClient().scanTable(tableName, null, segment,
      context.getSplit().getTotalSegments(), lastEvaluatedKey, lim.items, context.getReporter());

  ScanResult result = retryResult.result;
  int retries = retryResult.retries;

  double consumedCapacityUnits = 0.0;
  if (result.getConsumedCapacity() != null) {
    consumedCapacityUnits = result.getConsumedCapacity().getCapacityUnits();
  }
  return new PageResults<>(result.getItems(), result.getLastEvaluatedKey(), consumedCapacityUnits,
      retries);
}

private ScanResult getHashNumberRangeKeyItems(String[] hashKeys, String hashType) {
  List<Map<String, AttributeValue>> items = new ArrayList<>();
  for (String key : hashKeys) {
    for (Integer i = 0; i < NUM_RANGE_KEYS_PER_HASH_KEY; i++) {
      Map<String, AttributeValue> item = new HashMap<>();
      if (hashType.equals("S")) {
        item.put("hashKey", new AttributeValue(key));
      } else {
        item.put("hashKey", new AttributeValue().withN(key));
      }
      item.put("rangeKey", new AttributeValue().withN("0" + i.toString()));
      items.add(item);
    }
  }
  return new ScanResult().withScannedCount(items.size()).withItems(items).withConsumedCapacity
      (new ConsumedCapacity().withCapacityUnits(1d));
}

public ScanResult scan(final ScanRequest request, final int permitsToConsume) throws BackendException {
    setUserAgent(request);
    ScanResult result;
    timedReadThrottle(SCAN, request.getTableName(), permitsToConsume);

    final Timer.Context apiTimerContext = getTimerContext(SCAN, request.getTableName());
    try {
        result = client.scan(request);
    } catch (Exception e) {
        throw processDynamoDbApiException(e, SCAN, request.getTableName());
    } finally {
        apiTimerContext.stop();
    }
    meterConsumedCapacity(SCAN, result.getConsumedCapacity());
    measureItemCount(SCAN, request.getTableName(), result.getCount());
    return result;
}

@SuppressFBWarnings(value = "IT_NO_SUCH_ELEMENT",
    justification = "https://github.com/awslabs/dynamodb-janusgraph-storage-backend/issues/222")
@Override
public ScanResult next() {
    final Scan backoff = new Scan(request, delegate, lastConsumedCapacity);
    ScanResult result = null;
    try {
        result = backoff.runWithBackoff(); //this will be non-null or runWithBackoff throws
    } catch (BackendException e) {
        throw new BackendRuntimeException(e);
    }

    if (result.getConsumedCapacity() != null) {
        lastConsumedCapacity = result.getConsumedCapacity().getCapacityUnits().intValue();
    }

    if (result.getLastEvaluatedKey() != null && !result.getLastEvaluatedKey().isEmpty()) {
        hasNext = true;
        request.setExclusiveStartKey(result.getLastEvaluatedKey());
    } else {
        hasNext = false;
    }

    return result;
}

/**
 * Returns the list of usernames stored in the identity table.
 * 
 * @return list of existing usernames in DynamoDB table
 */
public List<String> listUsers() {
    List<String> users = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(USER_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        users.add(s);
    }

    return users;
}

/**
 * @return the list of device ID (UID) stored in the identity table.
 */
public List<String> listDevices() {
    List<String> devices = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(DEVICE_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        devices.add(s);
    }

    return devices;
}

/**
 * begins a scan with an exponential back off if throttled.
 */
public ScanResult runWithBackoff() {
    ScanResult result = null;
    boolean interrupted = false;
    try {
        do {
            try {
                result = client.scan(request);
            } catch (Exception e) {
                try {
                    Thread.sleep(exponentialBackoffTime);
                } catch (InterruptedException ie) {
                    interrupted = true;
                } finally {
                    exponentialBackoffTime *= 2;
                }
                continue;
            }
        } while (result == null);
        return result;
    } finally {
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    }
}

@Override
public List<LandUnit> getLandUnits(Long userId) {
    List<LandUnit> retval = new ArrayList<LandUnit>();
    try {
        /*
         * Scan items for movies with user id attribute.
         */
        Map<String, Condition> scanFilter = new HashMap<String, Condition>();
        Condition condition = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withN(userId.toString()));
        scanFilter.put(LandUnit.USER_ID_ATTR_NAME, condition);
        ScanRequest scanRequest =
                new ScanRequest(LANDUNIT_DYNAMO_DB_TABLE_NAME).withScanFilter(scanFilter);
        ScanResult scanResult = dynamoDB.scan(scanRequest);
        LOG.debug("DDB Scan Result: " + scanResult);
        retval = mapItemsToLandUnit(scanResult.getItems());
    } catch (Exception e) {
        LOG.error("Unable to retrieve land units from DDB " + e.getMessage());
    }
    return retval;
}

/**
 * Generate a list of attribute names found in the Aggregator's dynamo
 * table. Assumes that all Items in the Aggregator table are of the same
 * structure.
 * 
 * @param dynamoClient
 *            Dynamo DB Client to use for connection to Dynamo DB.
 * @param dynamoTable
 *            The Dynamo Table for the Aggregator
 * @return A list of attribute names from the Dynamo table
 * @throws Exception
 */
public static List<String> getDictionaryEntry(
        final AmazonDynamoDB dynamoClient, final String dynamoTable)
        throws Exception {
    // get a list of all columns in the table, with keys first
    List<String> columns = new ArrayList<>();
    List<KeySchemaElement> keys = dynamoClient.describeTable(dynamoTable)
            .getTable().getKeySchema();
    for (KeySchemaElement key : keys) {
        columns.add(key.getAttributeName());
    }
    ScanResult scan = dynamoClient.scan(new ScanRequest()
            .withTableName(dynamoTable).withSelect(Select.ALL_ATTRIBUTES)
            .withLimit(1));
    List<Map<String, AttributeValue>> scannedItems = scan.getItems();
    for (Map<String, AttributeValue> map : scannedItems) {
        for (String s : map.keySet()) {
            if (!columns.contains(s))
                columns.add(s);
        }
    }

    return columns;
}

/**
 * Generate a list of attribute names found in the Aggregator's dynamo
 * table. Assumes that all Items in the Aggregator table are of the same
 * structure.
 * 
 * @param dynamoClient Dynamo DB Client to use for connection to Dynamo DB.
 * @param dynamoTable The Dynamo Table for the Aggregator
 * @return A list of attribute names from the Dynamo table
 * @throws Exception
 */
protected List<String> getDictionaryEntry() throws Exception {
    // get a list of all columns in the table, with keys first
    List<String> columns = new ArrayList<>();
    List<KeySchemaElement> keys = dynamoClient.describeTable(this.tableName).getTable().getKeySchema();
    for (KeySchemaElement key : keys) {
        columns.add(key.getAttributeName());
    }
    ScanResult scan = dynamoClient.scan(new ScanRequest().withTableName(this.tableName).withSelect(
            Select.ALL_ATTRIBUTES).withLimit(1));
    List<Map<String, AttributeValue>> scannedItems = scan.getItems();
    for (Map<String, AttributeValue> map : scannedItems) {
        for (String s : map.keySet()) {
            if (!columns.contains(s))
                columns.add(s);
        }
    }

    return columns;
}

private void dumpTables() {
    for (String table : client.listTables().getTableNames()) {
        ScanResult scanResult;
        Map<String, AttributeValue> lastKey = null;
        do {
            scanResult = client.scan(new ScanRequest().withTableName(table).withExclusiveStartKey(lastKey));
            lastKey = scanResult.getLastEvaluatedKey();
            for (Map<String, AttributeValue> map : scanResult.getItems()) {
                for (Map.Entry<String, AttributeValue> item : map.entrySet()) {
                    System.out.print("item.put(\"");
                    System.out.print(item.getKey());
                    System.out.print("\", b642Av(\"");
                    System.out.print(Base64.encodeAsString(AttributeValueMarshaller.marshall(item.getValue()).array()));
                    System.out.println("\"));");
                }
                System.out.print("ddb.putItem(new PutItemRequest(\"");
                System.out.print(table);
                System.out.println("\", item));");
                System.out.println("item.clear();");
                System.out.println();
            }
        } while (lastKey != null);

    }
}

@Override
public List<LandUnit> getLandUnits(Long userId) {
    List<LandUnit> retval = new ArrayList<LandUnit>();
    try {
        /*
         * Scan items for movies with user id attribute.
         */
        Map<String, Condition> scanFilter = new HashMap<String, Condition>();
        Condition condition = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withN(userId.toString()));
        scanFilter.put(LandUnit.USER_ID_ATTR_NAME, condition);
        ScanRequest scanRequest =
                new ScanRequest(LANDUNIT_DYNAMO_DB_TABLE_NAME).withScanFilter(scanFilter);
        ScanResult scanResult = dynamoDB.scan(scanRequest);
        LOG.debug("DDB Scan Result: " + scanResult);
        retval = mapItemsToLandUnit(scanResult.getItems());
    } catch (Exception e) {
        LOG.error("Unable to retrieve land units from DDB " + e.getMessage());
    }
    return retval;
}

private static void findProductsForPriceLessThanZero() {

    Map<String, AttributeValue> expressionAttributeValues = new HashMap<String, AttributeValue>();
    expressionAttributeValues.put(":pr", new AttributeValue().withN("100"));

    ScanRequest scanRequest = new ScanRequest()
        .withTableName(tableName)
        .withFilterExpression("Price < :pr")
        .withExpressionAttributeValues(expressionAttributeValues)
        .withProjectionExpression("Id, Title, ProductCategory, Price");

    ScanResult result = client.scan(scanRequest);

    System.out.println("Scan of " + tableName + " for items with a price less than 100.");
    for (Map<String, AttributeValue> item : result.getItems()) {
        System.out.println("");
        printItem(item);
    }
}

/**
 * @return the list of device ID (UID) stored in the identity table.
 */
public List<String> listDevices() {
    List<String> devices = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(DEVICE_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        devices.add(s);
    }

    return devices;
}

/**
 * Returns the list of usernames stored in the identity table.
 * 
 * @return list of existing usernames in DynamoDB table
 */
public List<String> listUsers() {
    List<String> users = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(USER_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        users.add(s);
    }

    return users;
}

/**
 * @return the list of device ID (UID) stored in the identity table.
 */
public List<String> listDevices() {
    List<String> devices = new ArrayList<String>(1000);

    ScanResult result = ddb.scan(new ScanRequest().withTableName(DEVICE_TABLE).withLimit(1000));

    for (Map<String, AttributeValue> item : result.getItems()) {
        String s = "";

        for (Entry<String, AttributeValue> entry : item.entrySet()) {
            s += " ** " + entry.getKey() + " = " + entry.getValue().getS();
        }

        devices.add(s);
    }

    return devices;
}

public ScanResult getSince(long timestamp, int limit) throws Exception {
    try {
        Condition scanFilterCondition = new Condition()
                .withComparisonOperator(ComparisonOperator.GT.toString())
                .withAttributeValueList(
                        new AttributeValue().withN(Long.toString(timestamp)));
        Map<String, Condition> conditions = new HashMap<String, Condition>();
        conditions.put(COL_CREATEDAT, scanFilterCondition);

        ScanRequest scanRequest = new ScanRequest()
                .withTableName(TABLE_NAME)
                .withScanFilter(conditions)
                .withLimit(limit)
                .withAttributesToGet(
                        Arrays.asList(COL_ID, COL_CREATEDAT, COL_LAT,
                                COL_LONG, COL_SCREENNAME, COL_TEXT));

        return dynamoDB.scan(scanRequest);
    } catch (Exception e) {
        handleException(e);
    }

    return null;
}

private long getDynamoItemCount() {

        ScanResult scanResult = dynamoClient.scan(TEST_DYNAMO_TABLE_NAME, Lists.newArrayList("Uuid"));
        long itemCount = scanResult.getCount();

        while(scanResult.getLastEvaluatedKey() != null && !scanResult.getLastEvaluatedKey().isEmpty()) {
            scanResult = dynamoClient.scan(new ScanRequest().withTableName(TEST_DYNAMO_TABLE_NAME).withAttributesToGet("Uuid").withExclusiveStartKey(scanResult.getLastEvaluatedKey()));
            itemCount += scanResult.getCount();
        }

        return itemCount;
    }

/**
 * Count long.
 *
 * @return the long
 */
public long count() {
    final ScanRequest scan = new ScanRequest(TABLE_NAME);
    LOGGER.debug("Scanning table with request [{}] to count items", scan);
    final ScanResult result = this.amazonDynamoDBClient.scan(scan);
    LOGGER.debug("Scanned table with result [{}]", scan);
    return result.getCount();
}

/**
 * Gets all.
 *
 * @return the all
 */
public List<RegisteredService> getAll() {
    final List<RegisteredService> services = new ArrayList<>();
    final ScanRequest scan = new ScanRequest(TABLE_NAME);
    LOGGER.debug("Scanning table with request [{}]", scan);
    final ScanResult result = this.amazonDynamoDBClient.scan(scan);
    LOGGER.debug("Scanned table with result [{}]", scan);

    services.addAll(result.getItems()
            .stream()
            .map(this::deserializeServiceFromBinaryBlob)
            .sorted((o1, o2) -> Integer.valueOf(o1.getEvaluationOrder()).compareTo(o2.getEvaluationOrder()))
            .collect(Collectors.toList()));
    return services;
}

private Stream<Entry> scan(SecretEventStream.Filter<Entry> filter, Converters converters) {
    ScanRequest scanRequest = new ScanRequest();
    scanRequest.withConsistentRead(true);
    scanRequest.withTableName(tableName);

    FilterGenerator filterGenerator = new FilterGenerator();
    FilterGenerator.Filter generated = filterGenerator.process(filter.parsedAttributeCondition.get(), converters);

    if(!generated.expressionAttributeNames.isEmpty()) {
        scanRequest.withExpressionAttributeNames(generated.expressionAttributeNames);
    }

    if (!generated.expressionAttributeValues.isEmpty()) {
        scanRequest.withExpressionAttributeValues(generated.expressionAttributeValues);
    }

    scanRequest.withFilterExpression(generated.filterExpression);

    ScanResult result = client.scan(scanRequest);

    List<Map<String, AttributeValue>> results = new ArrayList<>();
    results.addAll(result.getItems());

    while (result.getLastEvaluatedKey() != null) {
        scanRequest = scanRequest.withExclusiveStartKey(result.getLastEvaluatedKey());

        result = client.scan(scanRequest);

        results.addAll(result.getItems());
    }

    Stream<Entry> typedResult = results.stream().map(this::fromMap);

    if (filter.reverse) {
        typedResult = Lists.reverse(typedResult.collect(Collectors.toCollection(LinkedList::new))).stream();
    }

    return typedResult;
}

@Test
public void testKeySet() throws Exception {
    ScanRequest request = new ScanRequest().withConsistentRead(true).withTableName(tableName);
    ScanResult result = constructScanResult();
    when(mockDynamoDBClient.scan(request)).thenReturn(result);

    // Call the KeySet method and assert the expected secret identifiers are returned.
    Set<SecretIdentifier> keys = dynamoDB.keySet();
    assertEquals(keys.size(), 2);
    assertTrue(keys.contains(new SecretIdentifier(SECRET_NAME)));
    assertTrue(keys.contains(new SecretIdentifier(SECRET2_NAME)));

    verify(mockDynamoDBClient, times(1)).scan(request);
}

@Test
public void testKeySetEmpty() throws Exception {
    ScanRequest request = new ScanRequest().withConsistentRead(true).withTableName(tableName);
    ScanResult result = new ScanResult().withCount(0).withItems(new ArrayList<>());
    when(mockDynamoDBClient.scan(request)).thenReturn(result);

    // Call the KeySet method and check that is it empty.
    Set<SecretIdentifier> keySet = dynamoDB.keySet();
    assertTrue(keySet.isEmpty());
    verify(mockDynamoDBClient, times(1)).scan(request);
}

@Test
public void fetchPageReturnsZeroConsumedCapacityWhenResultsConsumedCapacityIsNull() {
  RetryResult stubbedResult = new RetryResult<>(new ScanResult().withConsumedCapacity(null)
      .withItems(new HashMap<String, AttributeValue>()), 0);
  stubScanTableWith(stubbedResult);

  when(context.getClient()).thenReturn(client);
  when(context.getConf()).thenReturn(new JobConf());
  when(context.getSplit()).thenReturn(new DynamoDBSegmentsSplit());
  ScanReadManager readManager = Mockito.mock(ScanReadManager.class);
  ScanRecordReadRequest readRequest = new ScanRecordReadRequest(readManager, context, 0, null);
  PageResults<Map<String, AttributeValue>> pageResults =
      readRequest.fetchPage(new RequestLimit(0, 0));
  assertEquals(0.0, pageResults.consumedRcu, 0.0);
}

private void stubScanTableWith(RetryResult<ScanResult> scanResultRetryResult) {
  when(client.scanTable(
      anyString(),
      any(DynamoDBQueryFilter.class),
      anyInt(),
      anyInt(),
      any(Map.class),
      anyLong(),
      any(Reporter.class))
  ).thenReturn(scanResultRetryResult);
}

private ScanResult getHashKeyItems(String[] hashKeys, String type) {
  List<Map<String, AttributeValue>> items = new ArrayList<>();
  for (String key : hashKeys) {
    Map<String, AttributeValue> item = new HashMap<>();
    if (type.equals("S")) {
      item.put("hashKey", new AttributeValue(key));
    } else {
      item.put("hashKey", new AttributeValue().withN(key));
    }
    items.add(item);
  }
  return new ScanResult().withScannedCount(items.size()).withItems(items).withConsumedCapacity
      (new ConsumedCapacity().withCapacityUnits(1d));
}

private ScanOutcome scan(ScanSpec scanSpec, PageIterator pageIterator) {
    if ( null == _convertMarker ) {
        throw new IllegalStateException("Index must first be initialized with ConvertMarker");
    }
    if ( pageIterator.getPageSize() <= 0 ) {
        return new ScanOutcome(new ScanResult());
    }
    ItemCollection<ScanOutcome> itemCollection =
        maybeBackoff(true, () -> _scan.scan(withMarker(scanSpec, pageIterator)));

    if ( null != itemCollection ) {
        Iterator<Page<Item, ScanOutcome>> iterator = itemCollection.pages().iterator();
        if ( iterator.hasNext() ) {
            ScanOutcome outcome = maybeBackoff(true, () -> iterator.next().getLowLevelResult());
            ScanResult result = outcome.getScanResult();
            Map<String,AttributeValue> lastKey = result.getLastEvaluatedKey();
            if ( null != lastKey && ! lastKey.isEmpty() ) {
                pageIterator.setMarker(toMarker(lastKey, false));
            } else {
                pageIterator.setMarker(null);
            }
            return outcome;
        }
    }
    pageIterator.setMarker(null);
    return new ScanOutcome(new ScanResult());
}

@Override
public String handleRequest(Book request, Context context) {
    ScanRequest scanRequest = new ScanRequest().withTableName("Books");
    ScanResult result = DynamoDBUtil.getClient().scan(scanRequest);
    System.out.println("-- books listing start --");
    for (Map<String, AttributeValue> item : result.getItems()){
        System.out.println(item);
    }
    System.out.println("-- books listing end --");
    return result.getItems().toString();
}

@Override
public void execute() {
    ScanResult result = ddbClient.scan(new ScanRequest()
            .withTableName(determineTableName())
            .withScanFilter(determineScanFilter()));

    Map tmp = new HashMap<>();
    tmp.put(DdbConstants.ITEMS, result.getItems());
    tmp.put(DdbConstants.LAST_EVALUATED_KEY, result.getLastEvaluatedKey());
    tmp.put(DdbConstants.CONSUMED_CAPACITY, result.getConsumedCapacity());
    tmp.put(DdbConstants.COUNT, result.getCount());
    tmp.put(DdbConstants.SCANNED_COUNT, result.getScannedCount());
    addToResults(tmp);
}

@SuppressWarnings("unchecked")
@Override
public ScanResult scan(ScanRequest scanRequest) {
    this.scanRequest = scanRequest;
    ConsumedCapacity consumed = new ConsumedCapacity();
    consumed.setCapacityUnits(1.0);
    Map<String, AttributeValue> lastEvaluatedKey = new HashMap<String, AttributeValue>();
    lastEvaluatedKey.put("1", new AttributeValue("LAST_KEY"));
    return new ScanResult()
            .withConsumedCapacity(consumed)
            .withCount(1)
            .withItems(getAttributes())
            .withScannedCount(10)
            .withLastEvaluatedKey(lastEvaluatedKey);
}

@Override
public ScanContext call() throws Exception {
    try {
        final ScanRequest originalRequest = DynamoDbDelegate.copyScanRequest(request);
        final ScanResult result = next();
        return new ScanContext(originalRequest, result);
    } catch (BackendRuntimeException e) {
        throw e.getBackendException();
    }
}

@Override
public SegmentedScanResult call() {
    ScanResult result = null;
    result = runWithBackoff();

    final ConsumedCapacity cc = result.getConsumedCapacity();

    if (cc != null && cc.getCapacityUnits() != null) {
        lastConsumedCapacity = result.getConsumedCapacity()
                .getCapacityUnits().intValue();
    } else if (result.getScannedCount() != null && result.getCount() != null) {

        final boolean isConsistent = request.getConsistentRead();
        int itemSize = isConsistent ? BootstrapConstants.STRONGLY_CONSISTENT_READ_ITEM_SIZE
                : BootstrapConstants.EVENTUALLY_CONSISTENT_READ_ITEM_SIZE;

        lastConsumedCapacity = (result.getScannedCount() / (int) Math.max(1.0, result.getCount()))
                * (ItemSizeCalculator.calculateScanResultSizeInBytes(result) / itemSize);
    }

    if (result.getLastEvaluatedKey() != null
            && !result.getLastEvaluatedKey().isEmpty()) {
        hasNext = true;
        request.setExclusiveStartKey(result.getLastEvaluatedKey());
    } else {
        hasNext = false;
    }

    if (lastConsumedCapacity > 0) {
        rateLimiter.acquire(lastConsumedCapacity);
    }
    return new SegmentedScanResult(result, request.getSegment());
}

public static int calculateScanResultSizeInBytes(ScanResult result) {
    final Iterator<Map<String, AttributeValue>> it = result.getItems().iterator();
    int totalBytes = 0;
    while(it.hasNext()){
        totalBytes += calculateItemSizeInBytes(it.next());
    }
    return totalBytes;
}

@Override
public Void call() {
    final ScanResult scanResult = result.getScanResult();
    final List<Map<String, AttributeValue>> items = scanResult.getItems();
    final Iterator<Map<String, AttributeValue>> it = items.iterator();
    boolean interrupted = false;
    try {
        do {
            try {
                Map<String, AttributeValue> item = it.next();
                DynamoDBEntryWithSize entryWithSize = new DynamoDBEntryWithSize(
                        item,
                        ItemSizeCalculator.calculateItemSizeInBytes(item));
                queue.put(entryWithSize);
            } catch (InterruptedException e) {
                interrupted = true;
                LOGGER.warn("interrupted when writing item to queue: "
                        + e.getMessage());
            }
        } while (it.hasNext());
    } finally {
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    }
    return null;
}

/**
 * Splits up a ScanResult into a list of BatchWriteItemRequests of size 25
 * items or less each.
 */
public static List<BatchWriteItemRequest> splitResultIntoBatches(
        ScanResult result, String tableName) {
    List<BatchWriteItemRequest> batches = new LinkedList<BatchWriteItemRequest>();
    Iterator<Map<String, AttributeValue>> it = result.getItems().iterator();

    BatchWriteItemRequest req = new BatchWriteItemRequest()
            .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
    List<WriteRequest> writeRequests = new LinkedList<WriteRequest>();
    int i = 0;
    while (it.hasNext()) {
        PutRequest put = new PutRequest(it.next());
        writeRequests.add(new WriteRequest(put));

        i++;
        if (i == BootstrapConstants.MAX_BATCH_SIZE_WRITE_ITEM) {
            req.addRequestItemsEntry(tableName, writeRequests);
            batches.add(req);
            req = new BatchWriteItemRequest()
                    .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL);
            writeRequests = new LinkedList<WriteRequest>();
            i = 0;
        }
    }
    if (i > 0) {
        req.addRequestItemsEntry(tableName, writeRequests);
        batches.add(req);
    }
    return batches;
}

/**
 * Test the initialization of a BlockingQueueWorker and make sure it places the items in the queue when called.
 */
@Test
public void testInitializationAndCall() {
    ScanResult mockResult = createMock(ScanResult.class);
    SegmentedScanResult segmentedScanResult = new SegmentedScanResult(
            mockResult, 0);
    BlockingQueue<DynamoDBEntryWithSize> queue = new ArrayBlockingQueue<DynamoDBEntryWithSize>(
            20);
    BlockingQueueWorker callable = new BlockingQueueWorker(queue,
            segmentedScanResult);
    List<Map<String, AttributeValue>> items = new LinkedList<Map<String, AttributeValue>>();

    Map<String, AttributeValue> sampleScanResult = new HashMap<String, AttributeValue>();
    sampleScanResult.put("sample key", new AttributeValue(
            "sample attribute value"));
    items.add(sampleScanResult);

    expect(mockResult.getItems()).andReturn(items);

    replayAll();

    callable.call();

    verifyAll();

    assertEquals(1, queue.size());
    assertSame(sampleScanResult, queue.poll().getEntry());
}

/**
 * Test the getters and constructor of segmented scan result.
 */
@Test
public void test() {
    ScanResult result = new ScanResult();
    int numSegments = 3;
    SegmentedScanResult segmentedScanResult = new SegmentedScanResult(
            result, numSegments);

    assertSame(result, segmentedScanResult.getScanResult());
    assertEquals(numSegments, segmentedScanResult.getSegment());
}