You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by pp...@apache.org on 2021/08/05 07:18:22 UTC
[camel-quarkus] branch main updated: Expand AWS DDB tests #2776
This is an automated email from the ASF dual-hosted git repository.
ppalaga pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
The following commit(s) were added to refs/heads/main by this push:
new b04ea90 Expand AWS DDB tests #2776
b04ea90 is described below
commit b04ea90a4b69985d4f437662772c4cb630d575e7
Author: JiriOndrusek <on...@gmail.com>
AuthorDate: Wed Aug 4 10:15:16 2021 +0200
Expand AWS DDB tests #2776
---
.../component/aws2/ddb/it/Aws2DdbResource.java | 165 +++++++++++++-
.../aws2/ddb/it/Aws2DdbStreamResource.java | 43 +++-
.../component/aws2/ddb/it/Aws2DdbStreamRoutes.java | 36 +--
...source.java => TestSequenceNumberProvider.java} | 32 +--
.../component/aws2/ddb/it/Aws2DdbStreamIT.java} | 32 +--
.../component/aws2/ddb/it/Aws2DdbStreamTest.java | 245 +++++++++++++++++++++
.../quarkus/component/aws2/ddb/it/Aws2DdbTest.java | 230 ++++++++++++++-----
.../aws2/ddb/it/Aws2DdbTestEnvCustomizer.java | 73 ++++--
8 files changed, 701 insertions(+), 155 deletions(-)
diff --git a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbResource.java b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbResource.java
index f1aaddf..9ba1c51 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbResource.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbResource.java
@@ -21,7 +21,10 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
@@ -33,24 +36,41 @@ import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+import org.apache.camel.Message;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.component.aws2.ddb.Ddb2Constants;
import org.apache.camel.component.aws2.ddb.Ddb2Operations;
+import org.apache.camel.util.CollectionHelper;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import software.amazon.awssdk.services.dynamodb.model.AttributeAction;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import software.amazon.awssdk.services.dynamodb.model.AttributeValueUpdate;
+import software.amazon.awssdk.services.dynamodb.model.ComparisonOperator;
+import software.amazon.awssdk.services.dynamodb.model.Condition;
+import software.amazon.awssdk.services.dynamodb.model.KeysAndAttributes;
+import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughputDescription;
@Path("/aws2-ddb")
@ApplicationScoped
public class Aws2DdbResource {
+ public enum Table {
+ basic, operations, stream
+ }
+
@ConfigProperty(name = "aws-ddb.table-name")
String tableName;
+ @ConfigProperty(name = "aws-ddb.operations-table-name")
+ String operationsTableName;
+
+ @ConfigProperty(name = "aws-ddb.stream-table-name")
+ String streamTableName;
+
@Inject
ProducerTemplate producerTemplate;
@@ -59,7 +79,9 @@ public class Aws2DdbResource {
@POST
@Consumes(MediaType.TEXT_PLAIN)
@Produces(MediaType.TEXT_PLAIN)
- public Response post(String message, @PathParam("key") String key) throws Exception {
+ public Response post(String message,
+ @PathParam("key") String key,
+ @QueryParam("table") String table) throws Exception {
final Map<String, AttributeValue> item = new HashMap<String, AttributeValue>() {
{
put("key", AttributeValue.builder()
@@ -68,7 +90,8 @@ public class Aws2DdbResource {
.s(message).build());
}
};
- producerTemplate.sendBodyAndHeaders(componentUri(Ddb2Operations.PutItem),
+ producerTemplate.sendBodyAndHeaders(
+ componentUri(Table.valueOf(table), Ddb2Operations.PutItem),
message,
new HashMap<String, Object>() {
{
@@ -105,9 +128,9 @@ public class Aws2DdbResource {
@Path("/item/{key}")
@PUT
@Produces(MediaType.TEXT_PLAIN)
- public void updateItem(String message, @PathParam("key") String key) throws Exception {
+ public void updateItem(String message, @PathParam("key") String key, @QueryParam("table") String table) throws Exception {
producerTemplate.sendBodyAndHeaders(
- componentUri(Ddb2Operations.UpdateItem),
+ componentUri(Table.valueOf(table), Ddb2Operations.UpdateItem),
null,
new HashMap<String, Object>() {
{
@@ -130,9 +153,9 @@ public class Aws2DdbResource {
@Path("/item/{key}")
@DELETE
@Produces(MediaType.TEXT_PLAIN)
- public void deleteItem(@PathParam("key") String key) throws Exception {
+ public void deleteItem(@PathParam("key") String key, @QueryParam("table") String table) throws Exception {
producerTemplate.sendBodyAndHeaders(
- componentUri(Ddb2Operations.DeleteItem),
+ componentUri(Table.valueOf(table), Ddb2Operations.DeleteItem),
null,
new HashMap<String, Object>() {
{
@@ -144,8 +167,136 @@ public class Aws2DdbResource {
});
}
+ @Path("/batchItems")
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map batchItems(List<String> keyValues) throws Exception {
+ Map<String, AttributeValue>[] keyAttrs = keyValues.stream()
+ .map(v -> Collections.singletonMap("key", AttributeValue.builder().s(v).build())).toArray(Map[]::new);
+ Map<String, KeysAndAttributes> keysAttrs = Collections.singletonMap(operationsTableName,
+ KeysAndAttributes.builder().keys(keyAttrs).build());
+
+ Map<String, List<Map<AttributeValue, AttributeValue>>> result = (Map<String, List<Map<AttributeValue, AttributeValue>>>) producerTemplate
+ .send(componentUri(Table.operations, Ddb2Operations.BatchGetItems),
+ e -> e.getIn().setHeader(Ddb2Constants.BATCH_ITEMS, keysAttrs))
+ .getMessage().getHeader(Ddb2Constants.BATCH_RESPONSE);
+
+ Map<String, String> collected = new HashMap<>();
+ for (Map<AttributeValue, AttributeValue> m : result.get(operationsTableName)) {
+ collected.put(m.get("key").s(), m.get("value").s());
+ }
+ return collected;
+ }
+
+ @Path("/query")
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map query(String keyEq) throws Exception {
+ Map<String, Condition> keyConditions = new HashMap<>();
+ Condition.Builder condition = Condition.builder().comparisonOperator(ComparisonOperator.EQ.toString())
+ .attributeValueList(AttributeValue.builder().s(keyEq).build());
+
+ keyConditions.put("key", condition.build());
+
+ List<Map<AttributeValue, AttributeValue>> result = (List<Map<AttributeValue, AttributeValue>>) producerTemplate
+ .send(componentUri(Table.operations, Ddb2Operations.Query),
+ e -> {
+ e.getIn().setHeader(Ddb2Constants.ATTRIBUTE_NAMES,
+ Stream.of("key", "value").collect(Collectors.toList()));
+ e.getIn().setHeader(Ddb2Constants.CONSISTENT_READ, true);
+ e.getIn().setHeader(Ddb2Constants.LIMIT, 10);
+ e.getIn().setHeader(Ddb2Constants.SCAN_INDEX_FORWARD, true);
+ e.getIn().setHeader(Ddb2Constants.KEY_CONDITIONS, keyConditions);
+ })
+ .getMessage().getHeader(Ddb2Constants.ITEMS);
+
+ Map<String, String> collected = new HashMap<>();
+ for (Map<AttributeValue, AttributeValue> m : result) {
+ collected.put(m.get("key").s(), m.get("value").s());
+ }
+ return collected;
+ }
+
+ @Path("/scan")
+ @GET
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map scan() throws Exception {
+
+ List<Map<AttributeValue, AttributeValue>> result = (List<Map<AttributeValue, AttributeValue>>) producerTemplate
+ .send(componentUri(Table.operations, Ddb2Operations.Scan),
+ e -> {
+ e.getIn().setHeader(Ddb2Constants.ATTRIBUTE_NAMES,
+ Stream.of("key", "value").collect(Collectors.toList()));
+ e.getIn().setHeader(Ddb2Constants.CONSISTENT_READ, true);
+ })
+ .getMessage().getHeader(Ddb2Constants.ITEMS);
+
+ Map<String, String> collected = new HashMap<>();
+ for (Map<AttributeValue, AttributeValue> m : result) {
+ collected.put(m.get("key").s(), m.get("value").s());
+ }
+ return collected;
+ }
+
+ @Path("/updateTable")
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response updateTable(int capacity) throws Exception {
+ producerTemplate
+ .send(componentUri(Table.operations, Ddb2Operations.UpdateTable),
+ e -> {
+ e.getIn().setHeader(Ddb2Constants.READ_CAPACITY, capacity);
+ e.getIn().setHeader(Ddb2Constants.WRITE_CAPACITY, capacity);
+ });
+ return Response.created(new URI("https://camel.apache.org/")).build();
+ }
+
+ @Path("/operation")
+ @POST
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map operation(String operation) throws Exception {
+ final Message message = producerTemplate
+ .send(componentUri(Table.operations, Ddb2Operations.valueOf(operation)), e -> {
+ })
+ .getMessage();
+ return message.getHeaders().entrySet().stream().collect(Collectors.toMap(
+ Map.Entry::getKey,
+ e -> {
+ if (e.getValue() instanceof List) {
+ return ((List) e.getValue()).size();
+ }
+ if (e.getValue() instanceof ProvisionedThroughputDescription) {
+ ProvisionedThroughputDescription ptd = (ProvisionedThroughputDescription) e.getValue();
+ return CollectionHelper.mapOf(Ddb2Constants.READ_CAPACITY, ptd.readCapacityUnits(),
+ Ddb2Constants.WRITE_CAPACITY, ptd.writeCapacityUnits());
+ }
+ if (Ddb2Constants.TABLE_NAME.equals(e.getKey()) && operationsTableName.equals(e.getValue())) {
+ return Table.operations.toString();
+ }
+ return e.getValue() == null ? "" : e.getValue().toString();
+ }));
+ }
+
private String componentUri(Ddb2Operations op) {
- return "aws2-ddb://" + tableName + "?operation=" + op;
+ return componentUri(Table.basic, op);
}
+ private String componentUri(Table table, Ddb2Operations op) {
+ String tableName;
+
+ switch (table) {
+ case operations:
+ tableName = this.operationsTableName;
+ break;
+ case stream:
+ tableName = this.streamTableName;
+ break;
+ default:
+ tableName = this.tableName;
+ }
+ return "aws2-ddb://" + tableName + "?operation=" + op;
+ }
}
diff --git a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
index bb41701..27d890b 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
@@ -23,18 +23,24 @@ import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.inject.Named;
import javax.ws.rs.GET;
+import javax.ws.rs.POST;
import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.apache.camel.CamelContext;
@Path("/aws2-ddbstream")
@ApplicationScoped
public class Aws2DdbStreamResource {
- @ConfigProperty(name = "aws-ddb.table-name")
- String tableName;
+ @Inject
+ CamelContext camelContext;
+
+ @Inject
+ @Named("aws2DdbStreamSequenceNumberProvider")
+ TestSequenceNumberProvider sequenceNumberProvider;
@Inject
@Named("aws2DdbStreamReceivedEvents")
@@ -47,4 +53,35 @@ public class Aws2DdbStreamResource {
return aws2DdbStreamReceivedEvents;
}
+ @Path("/clear")
+ @GET
+ public void clear() {
+ aws2DdbStreamReceivedEvents.clear();
+ }
+
+ @Path("/setSequenceNumber")
+ @POST
+ public void setSequenceNumber(String newSn) {
+ sequenceNumberProvider.setLastSequenceNumber(newSn);
+ }
+
+ @GET
+ @Path("/route/{routeId}/{operation}")
+ @Produces(MediaType.TEXT_PLAIN)
+ public String route(@PathParam("routeId") String routeId, @PathParam("operation") String operation) throws Exception {
+ switch (operation) {
+ case "stop":
+ camelContext.getRouteController().stopRoute(routeId);
+ break;
+ case "start":
+ camelContext.getRouteController().startRoute(routeId);
+ break;
+ case "status":
+ return camelContext.getRouteController().getRouteStatus(routeId).name();
+
+ }
+
+ return null;
+ }
+
}
diff --git a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamRoutes.java b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamRoutes.java
index 41a4768..b7671d8 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamRoutes.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamRoutes.java
@@ -25,7 +25,6 @@ import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.inject.Named;
import javax.inject.Singleton;
-import javax.ws.rs.Produces;
import org.apache.camel.builder.RouteBuilder;
import org.eclipse.microprofile.config.inject.ConfigProperty;
@@ -35,8 +34,8 @@ import software.amazon.awssdk.services.dynamodb.model.StreamRecord;
@ApplicationScoped
public class Aws2DdbStreamRoutes extends RouteBuilder {
- @ConfigProperty(name = "aws-ddb.table-name")
- String tableName;
+ @ConfigProperty(name = "aws-ddb.stream-table-name")
+ String streamTableName;
@Inject
@Named("aws2DdbStreamReceivedEvents")
@@ -44,20 +43,23 @@ public class Aws2DdbStreamRoutes extends RouteBuilder {
@Override
public void configure() throws Exception {
- from("aws2-ddbstream://" + tableName)
- .process(e -> {
- Record record = e.getMessage().getBody(Record.class);
- StreamRecord item = record.dynamodb();
- Map<String, String> result = new LinkedHashMap<>();
- result.put("key", item.keys().get("key").s());
- if (item.hasOldImage()) {
- result.put("old", item.oldImage().get("value").s());
- }
- if (item.hasNewImage()) {
- result.put("new", item.newImage().get("value").s());
- }
- aws2DdbStreamReceivedEvents.add(result);
- });
+ from("aws2-ddbstream://" + streamTableName
+ + "?sequenceNumberProvider=#aws2DdbStreamSequenceNumberProvider&iteratorType=AT_SEQUENCE_NUMBER")
+ .id("aws2DdbStreamRoute")
+ .process(e -> {
+ Record record = e.getMessage().getBody(Record.class);
+ StreamRecord item = record.dynamodb();
+ Map<String, String> result = new LinkedHashMap<>();
+ result.put("key", item.keys().get("key").s());
+ if (item.hasOldImage()) {
+ result.put("old", item.oldImage().get("value").s());
+ }
+ if (item.hasNewImage()) {
+ result.put("new", item.newImage().get("value").s());
+ }
+ result.put("sequenceNumber", item.sequenceNumber());
+ aws2DdbStreamReceivedEvents.add(result);
+ });
}
static class Producers {
diff --git a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/TestSequenceNumberProvider.java
similarity index 59%
copy from integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
copy to integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/TestSequenceNumberProvider.java
index bb41701..e276715 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/TestSequenceNumberProvider.java
@@ -16,35 +16,23 @@
*/
package org.apache.camel.quarkus.component.aws2.ddb.it;
-import java.util.List;
-import java.util.Map;
-
import javax.enterprise.context.ApplicationScoped;
-import javax.inject.Inject;
import javax.inject.Named;
-import javax.ws.rs.GET;
-import javax.ws.rs.Path;
-import javax.ws.rs.Produces;
-import javax.ws.rs.core.MediaType;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.apache.camel.component.aws2.ddbstream.SequenceNumberProvider;
-@Path("/aws2-ddbstream")
@ApplicationScoped
-public class Aws2DdbStreamResource {
-
- @ConfigProperty(name = "aws-ddb.table-name")
- String tableName;
+@Named("aws2DdbStreamSequenceNumberProvider")
+public class TestSequenceNumberProvider implements SequenceNumberProvider {
- @Inject
- @Named("aws2DdbStreamReceivedEvents")
- List<Map<String, String>> aws2DdbStreamReceivedEvents;
+ private String lastSn = "0";
- @Path("/change")
- @GET
- @Produces(MediaType.APPLICATION_JSON)
- public List<Map<String, String>> change() {
- return aws2DdbStreamReceivedEvents;
+ @Override
+ public String getSequenceNumber() {
+ return lastSn;
}
+ public void setLastSequenceNumber(String newSn) {
+ lastSn = newSn;
+ }
}
diff --git a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamIT.java
similarity index 53%
copy from integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
copy to integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamIT.java
index bb41701..a740a00 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/main/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamResource.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamIT.java
@@ -16,35 +16,9 @@
*/
package org.apache.camel.quarkus.component.aws2.ddb.it;
-import java.util.List;
-import java.util.Map;
+import io.quarkus.test.junit.NativeImageTest;
-import javax.enterprise.context.ApplicationScoped;
-import javax.inject.Inject;
-import javax.inject.Named;
-import javax.ws.rs.GET;
-import javax.ws.rs.Path;
-import javax.ws.rs.Produces;
-import javax.ws.rs.core.MediaType;
-
-import org.eclipse.microprofile.config.inject.ConfigProperty;
-
-@Path("/aws2-ddbstream")
-@ApplicationScoped
-public class Aws2DdbStreamResource {
-
- @ConfigProperty(name = "aws-ddb.table-name")
- String tableName;
-
- @Inject
- @Named("aws2DdbStreamReceivedEvents")
- List<Map<String, String>> aws2DdbStreamReceivedEvents;
-
- @Path("/change")
- @GET
- @Produces(MediaType.APPLICATION_JSON)
- public List<Map<String, String>> change() {
- return aws2DdbStreamReceivedEvents;
- }
+@NativeImageTest
+class Aws2DdbStreamIT extends Aws2DdbStreamTest {
}
diff --git a/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamTest.java b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamTest.java
new file mode 100644
index 0000000..de0b7ab
--- /dev/null
+++ b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbStreamTest.java
@@ -0,0 +1,245 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.aws2.ddb.it;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import io.restassured.response.ExtractableResponse;
+import io.restassured.response.Response;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.quarkus.test.support.aws2.Aws2TestResource;
+import org.awaitility.Awaitility;
+import org.hamcrest.Matchers;
+import org.jboss.logging.Logger;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.quarkus.component.aws2.ddb.it.Aws2DdbResource.Table;
+
+@QuarkusTest
+@QuarkusTestResource(Aws2TestResource.class)
+class Aws2DdbStreamTest {
+
+ private static final Logger LOG = Logger.getLogger(Aws2DdbStreamTest.class);
+
+ @Test
+ public void stream() {
+ final String key1 = "key-1-" + UUID.randomUUID().toString().replace("-", "");
+ final String msg1 = "val-1" + UUID.randomUUID().toString().replace("-", "");
+ final String key2 = "key-2-" + UUID.randomUUID().toString().replace("-", "");
+ final String msg2 = "val-2-" + UUID.randomUUID().toString().replace("-", "");
+ final String key3 = "key-3-" + UUID.randomUUID().toString().replace("-", "");
+ final String msg3 = "val-3-" + UUID.randomUUID().toString().replace("-", "");
+
+ waitForStreamConsumerToStart();
+
+ /* Put #1 */
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(msg1)
+ .queryParam("table", Table.stream)
+ .post("/aws2-ddb/item/" + key1)
+ .then()
+ .statusCode(201);
+
+ /* Update #1 */
+ final String newMsg = "newVal" + UUID.randomUUID().toString().replace("-", "");
+ RestAssured.given()
+ .body(newMsg)
+ .queryParam("table", Table.stream)
+ .put("/aws2-ddb/item/" + key1)
+ .then()
+ .statusCode(204);
+
+ /* PUT #2 */
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(msg2)
+ .queryParam("table", Table.stream)
+ .post("/aws2-ddb/item/" + key2)
+ .then()
+ .statusCode(201);
+
+ StringBuilder put2SeqNumber = new StringBuilder();
+
+ Awaitility.await().atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.get("/aws2-ddbstream/change")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ List<Map> retVal = result.jsonPath().getList("$", Map.class);
+ LOG.info("Expecting 3 (+ 1 init) events got " + result.statusCode() + ": " + result.body().asString());
+ //remove init events
+ retVal = retVal.stream().filter(m -> !String.valueOf(m.get("key")).startsWith("initKey"))
+ .collect(Collectors.toList());
+
+ if (retVal.size() == 3 && retVal.get(2) != null && retVal.get(2).get("sequenceNumber") != null) {
+ put2SeqNumber.append(((String) retVal.get(2).get("sequenceNumber")));
+ }
+ return retVal;
+ },
+ /* The above actions should trigger the following three change events (initEvent is also present) */
+ list -> list.size() == 3
+
+ && key1.equals(list.get(0).get("key"))
+ && msg1.equals(list.get(0).get("new"))
+
+ && key1.equals(list.get(1).get("key"))
+ && msg1.equals(list.get(1).get("old"))
+ && newMsg.equals(list.get(1).get("new"))
+
+ && key2.equals(list.get(2).get("key"))
+ && msg2.equals(list.get(2).get("new"))
+
+ && put2SeqNumber.length() > 0);
+
+ /* SequenceNumber */
+
+ RestAssured.given()
+ .get("/aws2-ddbstream/clear")
+ .then()
+ .statusCode(204);
+
+ /* Restart route and clear results */
+ routeController("stop", null);
+ routeController("status", ServiceStatus.Stopped.name());
+
+ RestAssured.given()
+ .body(put2SeqNumber.toString())
+ .post("/aws2-ddbstream/setSequenceNumber")
+ .then()
+ .statusCode(204);
+
+ routeController("start", null);
+ routeController("status", ServiceStatus.Started.name());
+ waitForStreamConsumerToStart();
+
+ /* Put #3 */
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(msg3)
+ .queryParam("table", Table.stream)
+ .post("/aws2-ddb/item/" + key3)
+ .then()
+ .statusCode(201);
+
+ /* Delete #3 */
+ RestAssured.given()
+ .queryParam("table", Table.stream)
+ .delete("/aws2-ddb/item/" + key3)
+ .then()
+ .statusCode(204);
+
+ /* There should be only key2 and key3, because route is started with parameter "AT_SEQUENCE_NUMBER" of put #2. */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.get("/aws2-ddbstream/change")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ List<Map> retVal = result.jsonPath().getList("$", Map.class);
+
+ //remove init events
+ retVal = retVal.stream().filter(m -> !String.valueOf(m.get("key")).startsWith("initKey"))
+ .collect(Collectors.toList());
+
+ LOG.info("Expecting 3 events, got " + result.statusCode() + ": " + retVal);
+
+ return retVal;
+ },
+ list -> list.size() == 3
+
+ && key2.equals(list.get(0).get("key"))
+ && msg2.equals(list.get(0).get("new"))
+
+ && key3.equals(list.get(1).get("key"))
+ && msg3.equals(list.get(1).get("new"))
+
+ && key3.equals(list.get(2).get("key"))
+ && msg3.equals(list.get(2).get("old")));
+ }
+
+ private void waitForStreamConsumerToStart() {
+ final String initKeyPrefix = "initKey";
+ final String initMsg = "val";
+
+ /* Wait for the consumer to start. Test event has to be created periodically to ensure consumer reception */
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(initMsg)
+ .queryParam("table", Table.stream)
+ .post("/aws2-ddb/item/" + initKeyPrefix + UUID.randomUUID().toString().replace("-", ""))
+ .then()
+ .statusCode(201);
+
+ /* Periodically check that init event is received and if nothing is received, invoke another one */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.get("/aws2-ddbstream/change")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ LOG.info("Expecting at least 1 init event, got " + result.statusCode() + ": " + result.body().asString());
+
+ List<Map> res = result.jsonPath().getList("$", Map.class);
+
+ /* If there is no received event, consumer is still not running. Repeat insert. */
+ if (res.isEmpty()) {
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(initMsg)
+ .queryParam("table", Table.stream)
+ .post("/aws2-ddb/item/initKey" + UUID.randomUUID().toString().replace("-", ""))
+ .then()
+ .statusCode(201);
+ }
+ return res;
+ },
+ /* If at least one of the init events is recceived, consumer is working */
+ list -> !list.isEmpty());
+ }
+
+ private String routeController(String operation, String expectedResult) {
+ String routeId = "aws2DdbStreamRoute";
+ if (expectedResult == null) {
+ RestAssured.given()
+ .get("/aws2-ddbstream/route/" + routeId + "/" + operation)
+ .then().statusCode(204);
+ } else {
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).until(
+ () -> RestAssured
+ .get("/aws2-ddbstream/route/" + routeId + "/" + operation)
+ .then()
+ .statusCode(200)
+ .extract().asString(),
+ Matchers.is(expectedResult));
+ }
+
+ return null;
+ }
+}
diff --git a/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTest.java b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTest.java
index ad62d1d..60f46c5 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTest.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTest.java
@@ -16,11 +16,11 @@
*/
package org.apache.camel.quarkus.component.aws2.ddb.it;
-import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import io.quarkus.test.common.QuarkusTestResource;
import io.quarkus.test.junit.QuarkusTest;
@@ -28,11 +28,16 @@ import io.restassured.RestAssured;
import io.restassured.http.ContentType;
import io.restassured.response.ExtractableResponse;
import io.restassured.response.Response;
+import org.apache.camel.component.aws2.ddb.Ddb2Constants;
+import org.apache.camel.component.aws2.ddb.Ddb2Operations;
import org.apache.camel.quarkus.test.support.aws2.Aws2TestResource;
import org.awaitility.Awaitility;
import org.hamcrest.Matchers;
import org.jboss.logging.Logger;
import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.dynamodb.model.TableStatus;
+
+import static org.apache.camel.quarkus.component.aws2.ddb.it.Aws2DdbResource.Table;
@QuarkusTest
@QuarkusTestResource(Aws2TestResource.class)
@@ -44,42 +49,6 @@ class Aws2DdbTest {
public void crud() {
final String key = "key" + UUID.randomUUID().toString().replace("-", "");
final String msg = "val" + UUID.randomUUID().toString().replace("-", "");
- final String initKeyPrefix = "initKey";
- final String initMsg = "val";
-
- /* Wait for the consumer to start. Test event has to be created periodically to ensure consumer reception */
- RestAssured.given()
- .contentType(ContentType.TEXT)
- .body("")
- .post("/aws2-ddb/item/" + initKeyPrefix + UUID.randomUUID().toString().replace("-", ""))
- .then()
- .statusCode(201);
-
- /* Periodically check that init event is received and if nothing is received, invoke another one */
- Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
- () -> {
- ExtractableResponse<Response> result = RestAssured.get("/aws2-ddbstream/change")
- .then()
- .statusCode(200)
- .extract();
-
- LOG.info("Expecting at least 1 init event, got " + result.statusCode() + ": " + result.body().asString());
-
- List<Map> res = result.jsonPath().getList("$", Map.class);
-
- /* If there is no received event, consumer is still not running. Repeat insert. */
- if (res.isEmpty()) {
- RestAssured.given()
- .contentType(ContentType.TEXT)
- .body(initMsg)
- .post("/aws2-ddb/item/initKey" + UUID.randomUUID().toString().replace("-", ""))
- .then()
- .statusCode(201);
- }
- return res;
- },
- /* If at least one of the init events is recceived, consumer is working */
- list -> !list.isEmpty());
/* Ensure initially empty */
RestAssured.get("/aws2-ddb/item/" + key)
@@ -90,6 +59,7 @@ class Aws2DdbTest {
RestAssured.given()
.contentType(ContentType.TEXT)
.body(msg)
+ .queryParam("table", Table.basic)
.post("/aws2-ddb/item/" + key)
.then()
.statusCode(201);
@@ -108,6 +78,7 @@ class Aws2DdbTest {
/* Update */
final String newMsg = "newVal" + UUID.randomUUID().toString().replace("-", "");
RestAssured.given()
+ .queryParam("table", Table.basic)
.body(newMsg)
.put("/aws2-ddb/item/" + key)
.then()
@@ -125,6 +96,7 @@ class Aws2DdbTest {
/* Delete */
RestAssured.given()
+ .queryParam("table", Table.basic)
.delete("/aws2-ddb/item/" + key)
.then()
.statusCode(204);
@@ -139,31 +111,183 @@ class Aws2DdbTest {
},
Matchers.is(204));
- Awaitility.await().atMost(120, TimeUnit.SECONDS).until(
+ }
+
+ @Test
+ public void operations() {
+
+ final String key1 = "key-1-" + UUID.randomUUID().toString().replace("-", "");
+ final String msg1 = "val-1-" + UUID.randomUUID().toString().replace("-", "");
+ final String key2 = "key-2-" + UUID.randomUUID().toString().replace("-", "");
+ final String msg2 = "val-2-" + UUID.randomUUID().toString().replace("-", "");
+ final String key3 = "key-3-" + UUID.randomUUID().toString().replace("-", "");
+ final String msg3 = "val-3-" + UUID.randomUUID().toString().replace("-", "");
+
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .queryParam("table", Table.operations)
+ .body(msg1)
+ .post("/aws2-ddb/item/" + key1)
+ .then()
+ .statusCode(201);
+
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(msg2)
+ .queryParam("table", Table.operations)
+ .post("/aws2-ddb/item/" + key2)
+ .then()
+ .statusCode(201);
+
+ RestAssured.given()
+ .contentType(ContentType.TEXT)
+ .body(msg3)
+ .queryParam("table", Table.operations)
+ .post("/aws2-ddb/item/" + key3)
+ .then()
+ .statusCode(201);
+
+ /* Batch items */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+
+ ExtractableResponse<Response> result = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(Stream.of(key1, key2).collect(Collectors.toList()))
+ .post("/aws2-ddb/batchItems")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ LOG.info("Expecting 2 items, got " + result.statusCode() + ": " + result.body().asString());
+
+ return result.jsonPath().getMap("$");
+ },
+ /* Both inserted pairs have to be returned */
+ map -> map.size() == 2
+ && msg1.equals(map.get(key1))
+ && msg2.equals(map.get(key2)));
+
+ /* Query */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
() -> {
- ExtractableResponse<Response> result = RestAssured.get("/aws2-ddbstream/change")
+ ExtractableResponse<Response> result = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(key3)
+ .post("/aws2-ddb/query")
.then()
.statusCode(200)
.extract();
- LOG.info("Expecting 3 events got " + result.statusCode() + ": " + result.body().asString());
- List<Map> retVal = result.jsonPath().getList("$", Map.class);
- //remove init events
- return retVal.stream().filter(m -> !String.valueOf(m.get("key")).startsWith("initKey"))
- .collect(Collectors.toList());
+ LOG.info("Expecting 1 item, got " + result.statusCode() + ": " + result.body().asString());
+
+ return result.jsonPath().getMap("$");
},
- /* The above actions should trigger the following three change events (initEvent is also present) */
- list -> list.size() == 3
+ map -> map.size() == 1
+ && msg3.equals(map.get(key3)));
+
+ /* Scan */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.get("/aws2-ddb/scan")
+ .then()
+ .statusCode(200)
+ .extract();
- && key.equals(list.get(0).get("key"))
- && msg.equals(list.get(0).get("new"))
+ LOG.info("Expecting 3 items, got " + result.statusCode() + ": " + result.body().asString());
- && key.equals(list.get(1).get("key"))
- && msg.equals(list.get(1).get("old"))
- && newMsg.equals(list.get(1).get("new"))
+ return result.jsonPath().getMap("$");
+ },
+ map -> map.size() == 3
+ && msg1.equals(map.get(key1))
+ && msg2.equals(map.get(key2))
+ && msg3.equals(map.get(key3)));
- && key.equals(list.get(2).get("key"))
- && newMsg.equals(list.get(2).get("old")));
+ /* Describe table */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(Ddb2Operations.DescribeTable)
+ .post("/aws2-ddb/operation")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ LOG.info("Expecting table description, got " + result.statusCode() + ": " + result.body().asString());
+
+ return result.jsonPath().getMap("$");
+ },
+ map -> map.size() == 8
+ && map.containsKey(Ddb2Constants.CREATION_DATE)
+ && map.containsKey(Ddb2Constants.READ_CAPACITY)
+ && TableStatus.ACTIVE.name().equals(map.get(Ddb2Constants.TABLE_STATUS))
+ && map.containsKey(Ddb2Constants.WRITE_CAPACITY)
+ && map.containsKey(Ddb2Constants.TABLE_SIZE)
+ && map.containsKey(Ddb2Constants.KEY_SCHEMA)
+ && map.containsKey(Ddb2Constants.ITEM_COUNT)
+ && Table.operations == Table.valueOf((String) map.get(Ddb2Constants.TABLE_NAME)));
+
+ /* Update table */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(5)
+ .post("/aws2-ddb/updateTable")
+ .then()
+ .extract();
+
+ LOG.info("Expecting table update, got " + result.statusCode() + ": " + result.body().asString());
+
+ return result.statusCode();
+ },
+ Matchers.is(201));
+
+ /* Delete table (also verify that update from previous step took effect) */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(Ddb2Operations.DeleteTable)
+ .post("/aws2-ddb/operation")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ LOG.info("Expecting table deletion, got " + result.statusCode() + ": " + result.body().asString());
+
+ return result.jsonPath().getMap("$");
+ },
+ map -> map.size() == 7
+ && map.containsKey(Ddb2Constants.CREATION_DATE)
+ && map.containsKey(Ddb2Constants.TABLE_STATUS)
+ && map.containsKey(Ddb2Constants.TABLE_SIZE)
+ && map.containsKey(Ddb2Constants.KEY_SCHEMA)
+ && map.containsKey(Ddb2Constants.ITEM_COUNT)
+ && Table.operations == Table.valueOf((String) map.get(Ddb2Constants.TABLE_NAME))
+ //previous update changed throughput capacity from 10 to 5
+ && ((Map) map.get(Ddb2Constants.PROVISIONED_THROUGHPUT)).size() == 2
+ && ((Map) map.get(Ddb2Constants.PROVISIONED_THROUGHPUT)).get(Ddb2Constants.READ_CAPACITY).equals(5)
+ && ((Map) map.get(Ddb2Constants.PROVISIONED_THROUGHPUT)).get(Ddb2Constants.WRITE_CAPACITY).equals(5));
+
+ /* Verify delete with describe table */
+ Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(120, TimeUnit.SECONDS).until(
+ () -> {
+ ExtractableResponse<Response> result = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(Ddb2Operations.DescribeTable)
+ .post("/aws2-ddb/operation")
+ .then()
+ .statusCode(200)
+ .extract();
+
+ LOG.info("Expecting table description of non-existing table, got " + result.statusCode() + ": "
+ + result.body().asString());
+
+ return result.jsonPath().getMap("$");
+ },
+ map -> map.isEmpty());
}
}
diff --git a/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTestEnvCustomizer.java b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTestEnvCustomizer.java
index a2102c1..50c062c 100644
--- a/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTestEnvCustomizer.java
+++ b/integration-test-groups/aws2/aws2-ddb/src/test/java/org/apache/camel/quarkus/component/aws2/ddb/it/Aws2DdbTestEnvCustomizer.java
@@ -16,7 +16,10 @@
*/
package org.apache.camel.quarkus.component.aws2.ddb.it;
+import java.util.List;
import java.util.Locale;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import org.apache.camel.quarkus.test.support.aws2.Aws2TestEnvContext;
import org.apache.camel.quarkus.test.support.aws2.Aws2TestEnvCustomizer;
@@ -48,38 +51,60 @@ public class Aws2DdbTestEnvCustomizer implements Aws2TestEnvCustomizer {
final String tableName = "camel-quarkus-" + RandomStringUtils.randomAlphanumeric(16).toLowerCase(Locale.ROOT);
envContext.property("aws-ddb.table-name", tableName);
+ final String tableNameOperations = "camel-quarkus-operations-"
+ + RandomStringUtils.randomAlphanumeric(16).toLowerCase(Locale.ROOT);
+ envContext.property("aws-ddb.operations-table-name", tableNameOperations);
+
+ final String tableNameStreams = "camel-quarkus-streams-"
+ + RandomStringUtils.randomAlphanumeric(16).toLowerCase(Locale.ROOT);
+ envContext.property("aws-ddb.stream-table-name", tableNameStreams);
+
+ List<String> tableNames = Stream.of(tableName, tableNameStreams, tableNameOperations).collect(Collectors.toList());
+
final DynamoDbClient client = envContext.client(Service.DYNAMODB, DynamoDbClient::builder);
{
final String keyColumn = "key";
- client.createTable(
- CreateTableRequest.builder()
- .attributeDefinitions(AttributeDefinition.builder()
- .attributeName(keyColumn)
- .attributeType(ScalarAttributeType.S)
- .build())
- .keySchema(KeySchemaElement.builder()
- .attributeName(keyColumn)
- .keyType(KeyType.HASH)
- .build())
- .provisionedThroughput(ProvisionedThroughput.builder()
- .readCapacityUnits(new Long(10))
- .writeCapacityUnits(new Long(10))
- .build())
- .streamSpecification(StreamSpecification.builder()
- .streamEnabled(true)
- .streamViewType(StreamViewType.NEW_AND_OLD_IMAGES)
- .build())
- .tableName(tableName)
+
+ for (String table : tableNames) {
+ client.createTable(
+ createTableRequest(table, keyColumn)
+ .build());
+ }
+
+ for (String table : tableNames) {
+ try (DynamoDbWaiter dbWaiter = client.waiter()) {
+ dbWaiter.waitUntilTableExists(DescribeTableRequest.builder()
+ .tableName(table)
.build());
+ }
- try (DynamoDbWaiter dbWaiter = client.waiter()) {
- dbWaiter.waitUntilTableExists(DescribeTableRequest.builder()
- .tableName(tableName)
- .build());
+ envContext.closeable(() -> client.deleteTable(DeleteTableRequest.builder().tableName(table).build()));
}
- envContext.closeable(() -> client.deleteTable(DeleteTableRequest.builder().tableName(tableName).build()));
+ }
+ }
+
+ private CreateTableRequest.Builder createTableRequest(String tableName, String keyColumn) {
+ CreateTableRequest.Builder builder = CreateTableRequest.builder()
+ .attributeDefinitions(AttributeDefinition.builder()
+ .attributeName(keyColumn)
+ .attributeType(ScalarAttributeType.S)
+ .build())
+ .keySchema(KeySchemaElement.builder()
+ .attributeName(keyColumn)
+ .keyType(KeyType.HASH)
+ .build())
+ .provisionedThroughput(ProvisionedThroughput.builder()
+ .readCapacityUnits(new Long(10))
+ .writeCapacityUnits(new Long(10))
+ .build());
+ if (tableName.contains("streams")) {
+ builder.streamSpecification(StreamSpecification.builder()
+ .streamEnabled(true)
+ .streamViewType(StreamViewType.NEW_AND_OLD_IMAGES)
+ .build());
}
+ return builder.tableName(tableName);
}
}