Java 类com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClient 实例源码

@JsonIgnore
public void configure(AmazonDynamoDBAsyncClient client) {
  // set the region or the url, depending on the format
  if (regionOrUrl.contains(":")) {
    client.setEndpoint(regionOrUrl);
  } else {
    client.setRegion(RegionUtils.getRegion(regionOrUrl));
  }
}

private void ensureModel() {
  if (this.model == null) {
    this.client = new AmazonDynamoDBAsyncClient(config.inflateCredentials());
    config.getEndpoint().configure(client);
    this.model = new DynamoDB(client);
  }
}

@Override
protected void doStart() {
    dynamoDb = new AmazonDynamoDBAsyncClient(
            getCredentials(),
            getClientConfiguration(),
            AppenderExecutors.newExecutor(this, getThreadPoolSize())
    );
    dynamoDb.setRegion(RegionUtils.getRegion(region));
}

/**
 * Method to create the specific client to be used
 * 
 * @param clientType
 * @param credentials
 * @return
 */
public static AmazonDynamoDB getClient(String clientType,
    AWSCredentials credentials) {
  if (clientType.equals(SYNC_CLIENT_PROP))
    return new AmazonDynamoDBClient(credentials);
  if (clientType.equals(ASYNC_CLIENT_PROP))
    return new AmazonDynamoDBAsyncClient(credentials);
  return null;
}

public AmazonDynamoDBAsyncClient getDynamoClient()
{
    return dynamoClient;
}

public AmazonDynamoDBAsyncClient getAsyncClient() {
  return withProvider(new AmazonDynamoDBAsyncClient(credentials));
}

public AmazonDynamoDBAsyncClient getAsyncClient() {
  if (this.client == null) {
    this.client = getUtil().getAsyncClient();
  }
  return this.client;
}

@Override
protected Iterator<Page<Item, ?>> buildQuery(DynamoQueryBuilder builder,
  AmazonDynamoDBAsyncClient client) {
  return builder.withProps(scanProps).build(client).scan();
}

@Override
protected Iterator<Page<Item, ?>> buildQuery(DynamoQueryBuilder builder,
  AmazonDynamoDBAsyncClient client) {
  return builder.withProps(scanProps).build(client).query();
}

@Override
protected Iterator<Page<Item, ?>> buildQuery(DynamoQueryBuilder builder,
  AmazonDynamoDBAsyncClient client) {
  return builder.build(client).get();
}

public DynamoQuery build(AmazonDynamoDBAsyncClient client) {
  return new DynamoQuery(new DynamoDB(client).getTable(tableDef.getName()),
    columns == null || columns.size() == 0 ? "" : COMMAS.join(columns));
}

protected abstract Iterator<Page<Item, ?>> buildQuery(DynamoQueryBuilder builder,
AmazonDynamoDBAsyncClient client);

public static void cleanupAggTable(AWSCredentialsProvider credentials,
        Region region, final String dynamoTable, final String toSeq)
        throws Exception {
    final Double deleteBelow = Double.parseDouble(toSeq);

    // create two clients - one synchronous for the read of all candidate
    // values, and another for the delete operations
    final AmazonDynamoDB dynamoClient = new AmazonDynamoDBClient(
            credentials);
    if (region != null)
        dynamoClient.setRegion(region);
    final AmazonDynamoDBAsyncClient deleteCli = new AmazonDynamoDBAsyncClient(
            credentials);
    deleteCli.setRegion(region);
    Map<String, AttributeValue> lastKey = null;
    Map<String, AttributeValue> deleteKey = null;

    // work out what the key and date column name is
    String keyColumn = null;
    String dateColumn = null;

    List<KeySchemaElement> keySchema = dynamoClient
            .describeTable(dynamoTable).getTable().getKeySchema();
    for (KeySchemaElement element : keySchema) {
        if (element.getKeyType().equals(KeyType.HASH.name()))
            keyColumn = element.getAttributeName();

        if (element.getKeyType().equals(KeyType.RANGE.name()))
            dateColumn = element.getAttributeName();
    }

    LOG.info(String.format(
            "Deleting data from %s where %s values are below %s",
            dynamoTable, StreamAggregator.LAST_WRITE_SEQ, deleteBelow));
    int deleteCount = 0;

    do {
        // read data from the table
        ScanRequest scan = new ScanRequest()
                .withTableName(dynamoTable)
                .withAttributesToGet(keyColumn, dateColumn,
                        StreamAggregator.LAST_WRITE_SEQ)
                .withExclusiveStartKey(lastKey);

        ScanResult results = dynamoClient.scan(scan);

        // delete everything up to the system provided change number
        for (Map<String, AttributeValue> map : results.getItems()) {
            deleteKey = new HashMap<>();
            deleteKey.put(keyColumn, map.get(keyColumn));
            deleteKey.put(dateColumn, map.get(dateColumn));

            if (Double.parseDouble(map.get(StreamAggregator.LAST_WRITE_SEQ)
                    .getS()) < deleteBelow) {
                deleteCli.deleteItem(dynamoTable, deleteKey);
                deleteCount++;
            }
        }
        lastKey = results.getLastEvaluatedKey();
    } while (lastKey != null);

    LOG.info(String.format(
            "Operation Complete - %s Records removed from Aggregate Store",
            deleteCount));
}

public DynamoDataStore(AWSCredentialsProvider credentials, AggregatorType aggregatorType,
        String streamName, String tableName, String labelAttribute, String dateAttribute) {
    this(new AmazonDynamoDBAsyncClient(credentials), new AmazonKinesisClient(credentials),
            aggregatorType, streamName, tableName, labelAttribute, dateAttribute);
}