camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [6/7] camel git commit: Add an understanding of the shard list to the main dynamodbstream consumer.
Date Wed, 16 Dec 2015 13:22:51 GMT
Add an understanding of the shard list to the main dynamodbstream consumer.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ceaef105
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ceaef105
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ceaef105

Branch: refs/heads/master
Commit: ceaef105323d031ef92751d2744ad7a0f0be2c08
Parents: fdadd15
Author: Candle <candle@candle.me.uk>
Authored: Fri Dec 11 16:25:48 2015 +0000
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Wed Dec 16 14:19:12 2015 +0100

----------------------------------------------------------------------
 .../component/aws/ddbstream/DdbStreamConsumer.java   | 15 ++++++++++++++-
 .../camel/component/aws/ddbstream/ShardList.java     | 11 +++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/ceaef105/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
index 88d2ba5..d4c9ac7 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumer.java
@@ -26,6 +26,7 @@ import com.amazonaws.services.dynamodbv2.model.GetShardIteratorResult;
 import com.amazonaws.services.dynamodbv2.model.ListStreamsRequest;
 import com.amazonaws.services.dynamodbv2.model.ListStreamsResult;
 import com.amazonaws.services.dynamodbv2.model.Record;
+import com.amazonaws.services.dynamodbv2.model.Shard;
 import java.util.ArrayDeque;
 import java.util.List;
 import java.util.Queue;
@@ -42,6 +43,8 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer {
     private static final Logger LOG = LoggerFactory.getLogger(DdbStreamConsumer.class);
 
     private String currentShardIterator = null;
+    private Shard currentShard;
+    private ShardList shardList = new ShardList();
 
     public DdbStreamConsumer(DdbStreamEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -105,10 +108,20 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer
{
                     .withStreamArn(streamArn)
                     ;
             DescribeStreamResult res1 = getClient().describeStream(req1);
+            shardList.addAll(res1.getStreamDescription().getShards());
+
+            LOG.trace("Current shard is: {} (in {})", currentShard, shardList);
+            if (currentShard == null) {
+                currentShard = shardList.first();
+            } else {
+                currentShard = shardList.nextAfter(currentShard);
+            }
+            shardList.removeOlderThan(currentShard);
+            LOG.trace("Next shard is: {} (in {})", currentShard, shardList);
 
             GetShardIteratorRequest req = new GetShardIteratorRequest()
                     .withStreamArn(streamArn)
-                    .withShardId(res1.getStreamDescription().getShards().get(0).getShardId())
// XXX only uses the first shard
+                    .withShardId(currentShard.getShardId())
                     .withShardIteratorType(getEndpoint().getIteratorType())
                     ;
             GetShardIteratorResult result = getClient().getShardIterator(req);

http://git-wip-us.apache.org/repos/asf/camel/blob/ceaef105/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardList.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardList.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardList.java
index fb188f4..a85e703 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardList.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardList.java
@@ -4,8 +4,11 @@ import com.amazonaws.services.dynamodbv2.model.Shard;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 class ShardList {
+    private final Logger LOG = LoggerFactory.getLogger(ShardList.class);
 
     private final Map<String, Shard> shards = new HashMap<>();
 
@@ -45,13 +48,21 @@ class ShardList {
     void removeOlderThan(Shard removeBefore) {
         String current = removeBefore.getParentShardId();
 
+        int removedShards = 0;
         while (current != null) {
             Shard s = shards.remove(current);
             if (s == null) {
                 current = null;
             } else {
+                removedShards++;
                 current = s.getParentShardId();
             }
         }
+        LOG.trace("removed {} shards from the store, new size is {}", removedShards, shards.size());
+    }
+
+    @Override
+    public String toString() {
+        return "ShardList{" + "shards=" + shards + '}';
     }
 }
\ No newline at end of file


Mime
View raw message