camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/9] camel git commit: CAMEL-9515 refactor to extract the shard iterator discovery code, as it's going to get more complicated for this ticket.
Date Thu, 21 Jan 2016 09:14:54 GMT
Repository: camel
Updated Branches:
  refs/heads/master e23bf831a -> 2f1c8599c


CAMEL-9515 refactor to extract the shard iterator discovery code, as it's going to get more
complicated for this ticket.


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

Branch: refs/heads/master
Commit: d097871c664f17fad96c19eb014fd6bd828d1357
Parents: 81c5a19
Author: Candle <candle@candle.me.uk>
Authored: Mon Jan 18 09:48:11 2016 +0000
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Thu Jan 21 10:10:24 2016 +0100

----------------------------------------------------------------------
 .../aws/ddbstream/DdbStreamConsumer.java        |  86 +--------------
 .../aws/ddbstream/ShardIteratorHandler.java     | 110 +++++++++++++++++++
 2 files changed, 114 insertions(+), 82 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/d097871c/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 aad6550..4bdb5c6 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
@@ -22,17 +22,9 @@ import java.util.List;
 import java.util.Queue;
 
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams;
-import com.amazonaws.services.dynamodbv2.model.DescribeStreamRequest;
-import com.amazonaws.services.dynamodbv2.model.DescribeStreamResult;
 import com.amazonaws.services.dynamodbv2.model.GetRecordsRequest;
 import com.amazonaws.services.dynamodbv2.model.GetRecordsResult;
-import com.amazonaws.services.dynamodbv2.model.GetShardIteratorRequest;
-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 com.amazonaws.services.dynamodbv2.model.ShardIteratorType;
 import org.apache.camel.AsyncCallback;
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
@@ -45,25 +37,24 @@ import org.slf4j.LoggerFactory;
 public class DdbStreamConsumer extends ScheduledBatchPollingConsumer {
     private static final Logger LOG = LoggerFactory.getLogger(DdbStreamConsumer.class);
 
-    private String currentShardIterator;
-    private Shard currentShard;
-    private final ShardList shardList = new ShardList();
+    private final ShardIteratorHandler shardIteratorHandler;
 
     public DdbStreamConsumer(DdbStreamEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
+        shardIteratorHandler = new ShardIteratorHandler(endpoint);
     }
 
     @Override
     protected int poll() throws Exception {
         GetRecordsRequest req = new GetRecordsRequest()
-                .withShardIterator(getShardIterator())
+                .withShardIterator(shardIteratorHandler.getShardIterator())
                 .withLimit(getEndpoint().getMaxResultsPerRequest());
         GetRecordsResult result = getClient().getRecords(req);
 
         Queue<Exchange> exchanges = createExchanges(result.getRecords());
         int processedExchangeCount = processBatch(CastUtils.cast(exchanges));
 
-        currentShardIterator = result.getNextShardIterator();
+        shardIteratorHandler.updateShardIterator(result.getNextShardIterator());
 
         return processedExchangeCount;
     }
@@ -95,75 +86,6 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer {
         return (DdbStreamEndpoint) super.getEndpoint();
     }
 
-    private String getShardIterator() {
-        // either return a cached one or get a new one via a GetShardIterator request.
-        if (currentShardIterator == null) {
-            ListStreamsRequest req0 = new ListStreamsRequest()
-                    .withTableName(getEndpoint().getTableName());
-            ListStreamsResult res0 = getClient().listStreams(req0);
-            final String streamArn = res0.getStreams().get(0).getStreamArn(); // XXX assumes
there is only one stream
-            DescribeStreamRequest req1 = new DescribeStreamRequest()
-                    .withStreamArn(streamArn);
-            DescribeStreamResult res1 = getClient().describeStream(req1);
-            shardList.addAll(res1.getStreamDescription().getShards());
-
-            LOG.trace("Current shard is: {} (in {})", currentShard, shardList);
-            if (currentShard == null) {
-                switch(getEndpoint().getIteratorType()) {
-                case AFTER_SEQUENCE_NUMBER:
-                    currentShard = shardList.afterSeq(getEndpoint().getSequenceNumber());
-                    break;
-                case AT_SEQUENCE_NUMBER:
-                    currentShard = shardList.atSeq(getEndpoint().getSequenceNumber());
-                    break;
-                case TRIM_HORIZON:
-                    currentShard = shardList.first();
-                    break;
-                case LATEST:
-                default:
-                    currentShard = shardList.last();
-                    break;
-                }
-            } else {
-                currentShard = shardList.nextAfter(currentShard);
-            }
-            shardList.removeOlderThan(currentShard);
-            LOG.trace("Next shard is: {} (in {})", currentShard, shardList);
-
-            GetShardIteratorRequest req = new GetShardIteratorRequest()
-                    .withStreamArn(streamArn)
-                    .withShardId(currentShard.getShardId())
-                    .withShardIteratorType(getEndpoint().getIteratorType());
-            switch(getEndpoint().getIteratorType()) {
-            case AFTER_SEQUENCE_NUMBER:
-            case AT_SEQUENCE_NUMBER:
-                // if you request with a sequence number that is LESS than the
-                // start of the shard, you get a HTTP 400 from AWS.
-                // So only add the sequence number if the endpoints
-                // sequence number is less than or equal to the starting
-                // sequence for the shard.
-                // Otherwise change the shart iterator type to trim_horizon
-                // because we get a 400 when we use one of the
-                // {at,after}_sequence_number iterator types and don't supply
-                // a sequence number.
-                if (BigIntComparisons.Conditions.LTEQ.matches(
-                        new BigInteger(currentShard.getSequenceNumberRange().getStartingSequenceNumber()),
-                        new BigInteger(getEndpoint().getSequenceNumber())
-                )) {
-                    req = req.withSequenceNumber(getEndpoint().getSequenceNumber());
-                } else {
-                    req = req.withShardIteratorType(ShardIteratorType.TRIM_HORIZON);
-                }
-                break;
-            default:
-            }
-            GetShardIteratorResult result = getClient().getShardIterator(req);
-            currentShardIterator = result.getShardIterator();
-        }
-        LOG.trace("Shard Iterator is: {}", currentShardIterator);
-        return currentShardIterator;
-    }
-
     private Queue<Exchange> createExchanges(List<Record> records) {
         Queue<Exchange> exchanges = new ArrayDeque<>();
         BigIntComparisons condition;

http://git-wip-us.apache.org/repos/asf/camel/blob/d097871c/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
new file mode 100644
index 0000000..42c111d
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/ShardIteratorHandler.java
@@ -0,0 +1,110 @@
+package org.apache.camel.component.aws.ddbstream;
+
+import java.math.BigInteger;
+
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams;
+import com.amazonaws.services.dynamodbv2.model.DescribeStreamRequest;
+import com.amazonaws.services.dynamodbv2.model.DescribeStreamResult;
+import com.amazonaws.services.dynamodbv2.model.GetShardIteratorRequest;
+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.Shard;
+import com.amazonaws.services.dynamodbv2.model.ShardIteratorType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class ShardIteratorHandler {
+    private static final Logger LOG = LoggerFactory.getLogger(ShardIteratorHandler.class);
+
+    private final DdbStreamEndpoint endpoint;
+    private final ShardList shardList = new ShardList();
+
+    private String currentShardIterator;
+    private Shard currentShard;
+
+    ShardIteratorHandler(DdbStreamEndpoint endpoint) {
+        this.endpoint = endpoint;
+    }
+
+    String getShardIterator() {
+        // either return a cached one or get a new one via a GetShardIterator request.
+        if (currentShardIterator == null) {
+            ListStreamsRequest req0 = new ListStreamsRequest()
+                    .withTableName(getEndpoint().getTableName());
+            ListStreamsResult res0 = getClient().listStreams(req0);
+            final String streamArn = res0.getStreams().get(0).getStreamArn(); // XXX assumes
there is only one stream
+            DescribeStreamRequest req1 = new DescribeStreamRequest()
+                    .withStreamArn(streamArn);
+            DescribeStreamResult res1 = getClient().describeStream(req1);
+            shardList.addAll(res1.getStreamDescription().getShards());
+
+            LOG.trace("Current shard is: {} (in {})", currentShard, shardList);
+            if (currentShard == null) {
+                switch(getEndpoint().getIteratorType()) {
+                case AFTER_SEQUENCE_NUMBER:
+                    currentShard = shardList.afterSeq(getEndpoint().getSequenceNumber());
+                    break;
+                case AT_SEQUENCE_NUMBER:
+                    currentShard = shardList.atSeq(getEndpoint().getSequenceNumber());
+                    break;
+                case TRIM_HORIZON:
+                    currentShard = shardList.first();
+                    break;
+                case LATEST:
+                default:
+                    currentShard = shardList.last();
+                    break;
+                }
+            } else {
+                currentShard = shardList.nextAfter(currentShard);
+            }
+            shardList.removeOlderThan(currentShard);
+            LOG.trace("Next shard is: {} (in {})", currentShard, shardList);
+
+            GetShardIteratorRequest req = new GetShardIteratorRequest()
+                    .withStreamArn(streamArn)
+                    .withShardId(currentShard.getShardId())
+                    .withShardIteratorType(getEndpoint().getIteratorType());
+            switch(getEndpoint().getIteratorType()) {
+            case AFTER_SEQUENCE_NUMBER:
+            case AT_SEQUENCE_NUMBER:
+                // if you request with a sequence number that is LESS than the
+                // start of the shard, you get a HTTP 400 from AWS.
+                // So only add the sequence number if the endpoints
+                // sequence number is less than or equal to the starting
+                // sequence for the shard.
+                // Otherwise change the shart iterator type to trim_horizon
+                // because we get a 400 when we use one of the
+                // {at,after}_sequence_number iterator types and don't supply
+                // a sequence number.
+                if (BigIntComparisons.Conditions.LTEQ.matches(
+                        new BigInteger(currentShard.getSequenceNumberRange().getStartingSequenceNumber()),
+                        new BigInteger(getEndpoint().getSequenceNumber())
+                )) {
+                    req = req.withSequenceNumber(getEndpoint().getSequenceNumber());
+                } else {
+                    req = req.withShardIteratorType(ShardIteratorType.TRIM_HORIZON);
+                }
+                break;
+            default:
+            }
+            GetShardIteratorResult result = getClient().getShardIterator(req);
+            currentShardIterator = result.getShardIterator();
+        }
+        LOG.trace("Shard Iterator is: {}", currentShardIterator);
+        return currentShardIterator;
+    }
+
+    void updateShardIterator(String nextShardIterator) {
+        this.currentShardIterator = nextShardIterator;
+    }
+
+    DdbStreamEndpoint getEndpoint() {
+        return endpoint;
+    }
+   
+    private AmazonDynamoDBStreams getClient() {
+        return getEndpoint().getClient();
+    }
+}


Mime
View raw message