package com.reactive.dynamodb;

import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
import com.amazonaws.services.dynamodbv2.model.*;
import rx.Observable;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
import static com.reactive.dynamodb.GetItemRequestFactory.make;

class ReactiveDynamo {
    private final AmazonDynamoDBAsync db;

    ReactiveDynamo(AWSCredentialsProvider awsCredentialsProvider, EndpointConfiguration endpointConfiguration) {
        this.db = AmazonDynamoDBAsyncClientBuilder.standard()
                .withEndpointConfiguration(endpointConfiguration)
                .withCredentials(awsCredentialsProvider)
                .build();
    }

    Observable<List<String>> tables() {
        return Observable.from(db.listTablesAsync())
                .map(ListTablesResult::getTableNames);
    }

    Observable<Map<String, Object>> item(DynamoDbTable table) {
        return Observable.from(db.getItemAsync(make(table)))
                .map(GetItemResult::getItem)
                .map(AwsInternalsExtension::toSimpleMapValue);
    }

    Observable<Map<String, Object>> add(String tableName, Map<String, Object> data) {
        return Observable.from(db.putItemAsync(PutItemRequestFactory.make(tableName, data)))
                .map(PutItemResult::getAttributes)
                .map(AwsInternalsExtension::toSimpleMapValue);
    }

    Observable<List<Map<String, Object>>> items(DynamoDbTable table) {
        Observable<QueryResult> observable = Observable.from(db.queryAsync(QueryRequestFactory.make(table)));
        return observable.mergeWith(observable.flatMap((QueryResult r) -> items(table, r.getLastEvaluatedKey())))
                .map(QueryResult::getItems)
                .map(items -> items.stream()
                        .map(AwsInternalsExtension::toSimpleMapValue)
                        .collect(Collectors.toList()));
    }

    private Observable<QueryResult> items(DynamoDbTable table, Map<String, AttributeValue> lastEvaluatedKey) {
        if (lastEvaluatedKey == null || lastEvaluatedKey.size() == 0) return Observable.empty();
        QueryRequest queryRequest = QueryRequestFactory.make(table, lastEvaluatedKey);
        Observable<QueryResult> observable = Observable.from(db.queryAsync(queryRequest));
        return observable.mergeWith(observable.flatMap((QueryResult r) -> items(table, r.getLastEvaluatedKey())));
    }
}

@@ -1690,221 +1690,64 @@
     
-Observable%3CQueryResult%3E observable = Observable.from(db.queryAsync(QueryRequestFactory.make(table)));%0A        return observable.mergeWith(observable.flatMap((QueryResult r) -%3E items(table, r.getLastEvaluatedKey())
+return mergedItems(table, QueryRequestFactory.make(table
 ))%0A 
@@ -2187,35 +2187,33 @@
     
-QueryRequest queryRequest =
+return mergedItems(table,
  Que
@@ -2258,17 +2258,123 @@
 atedKey)
-;
+);%0A    %7D%0A%0A    private Observable%3CQueryResult%3E mergedItems(DynamoDbTable table, QueryRequest queryRequest) %7B
 %0A       

Remove duplication