camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [5/6] camel git commit: Add filtering of the records when asking for an {at, after}_sequence_number iterator type.
Date Thu, 24 Dec 2015 08:46:08 GMT
Add filtering of the records when asking for an {at,after}_sequence_number iterator type.


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

Branch: refs/heads/master
Commit: 8e05657d6a742bb6b8b0026c9683953c38168a66
Parents: b2430c0
Author: Candle <candle@candle.me.uk>
Authored: Mon Dec 21 18:29:38 2015 +0000
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Thu Dec 24 09:45:32 2015 +0100

----------------------------------------------------------------------
 .../aws/ddbstream/AtAfterCondition.java         | 44 +++++++++++++
 .../aws/ddbstream/DdbStreamConsumer.java        | 45 ++++++++++++-
 .../component/aws/ddbstream/ShardList.java      | 24 +------
 .../aws/ddbstream/AtAfterConditionTest.java     | 67 ++++++++++++++++++++
 .../aws/ddbstream/DdbStreamConsumerTest.java    | 56 +++++++++++++++-
 5 files changed, 212 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/8e05657d/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/AtAfterCondition.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/AtAfterCondition.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/AtAfterCondition.java
new file mode 100644
index 0000000..a6798b5
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/ddbstream/AtAfterCondition.java
@@ -0,0 +1,44 @@
+/**
+ * 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.component.aws.ddbstream;
+
+import java.math.BigInteger;
+
+interface AtAfterCondition {
+
+    /**
+     * @return true if sequenceNumber is (at,after) the endpointSequenceNumber.
+     */
+    boolean matches(BigInteger endpointSequenceNumber, BigInteger sequenceNumber);
+
+    static enum Conditions implements AtAfterCondition {
+        AFTER() {
+            @Override
+            public boolean matches(BigInteger endpointSequenceNumber, BigInteger sequenceNumber)
{
+                return endpointSequenceNumber.compareTo(sequenceNumber) < 0;
+            }
+        },
+
+        AT() {
+            @Override
+            public boolean matches(BigInteger endpointSequenceNumber, BigInteger sequenceNumber)
{
+                return endpointSequenceNumber.compareTo(sequenceNumber) <= 0;
+            }
+        }
+        // TODO rename to LT/LTEQ/EQ/GTEQ/GT
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/8e05657d/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 25e5f31..d520abf 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
@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.aws.ddbstream;
 
+import java.math.BigInteger;
 import java.util.ArrayDeque;
 import java.util.List;
 import java.util.Queue;
@@ -31,6 +32,7 @@ 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;
@@ -132,6 +134,30 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer
{
                     .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 AT or AFTER 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 (AtAfterCondition.Conditions.AT.matches(
+                        new BigInteger(currentShard.getSequenceNumberRange().getStartingSequenceNumber()),
+                        new BigInteger(getEndpoint().getSequenceNumber())
+                )) {
+                    req = req.withSequenceNumber(getEndpoint().getSequenceNumber())
+                        .withShardIteratorType(getEndpoint().getIteratorType());
+                } else {
+                    req = req.withShardIteratorType(ShardIteratorType.TRIM_HORIZON);
+                }
+                break;
+            default:
+            }
             GetShardIteratorResult result = getClient().getShardIterator(req);
             currentShardIterator = result.getShardIterator();
         }
@@ -141,8 +167,25 @@ public class DdbStreamConsumer extends ScheduledBatchPollingConsumer
{
 
     private Queue<Exchange> createExchanges(List<Record> records) {
         Queue<Exchange> exchanges = new ArrayDeque<>();
+        AtAfterCondition condition;
+        BigInteger providedSeqNum = null;
+        switch(getEndpoint().getIteratorType()) {
+        case AFTER_SEQUENCE_NUMBER:
+            condition = AtAfterCondition.Conditions.AFTER;
+            providedSeqNum = new BigInteger(getEndpoint().getSequenceNumberProvider().getSequenceNumber());
+            break;
+        case AT_SEQUENCE_NUMBER:
+            condition = AtAfterCondition.Conditions.AT;
+            providedSeqNum = new BigInteger(getEndpoint().getSequenceNumberProvider().getSequenceNumber());
+            break;
+        default:
+            condition = null;
+        }
         for (Record record : records) {
-            exchanges.add(getEndpoint().createExchange(record));
+            BigInteger recordSeqNum = new BigInteger(record.getDynamodb().getSequenceNumber());
+            if (condition == null || condition.matches(providedSeqNum, recordSeqNum)) {
+                exchanges.add(getEndpoint().createExchange(record));
+            }
         }
         return exchanges;
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/8e05657d/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 0a6e332..5c5bd29 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
@@ -79,11 +79,11 @@ class ShardList {
     }
 
     Shard afterSeq(String sequenceNumber) {
-        return atAfterSeq(sequenceNumber, After.INSTANCE);
+        return atAfterSeq(sequenceNumber, AtAfterCondition.Conditions.AFTER);
     }
 
     Shard atSeq(String sequenceNumber) {
-        return atAfterSeq(sequenceNumber, At.INSTANCE);
+        return atAfterSeq(sequenceNumber, AtAfterCondition.Conditions.AT);
     }
 
     Shard atAfterSeq(String sequenceNumber, AtAfterCondition condition) {
@@ -134,27 +134,9 @@ class ShardList {
         return "ShardList{" + "shards=" + shards + '}';
     }
 
-    private interface AtAfterCondition {
-        boolean matches(BigInteger sequenceNumber, BigInteger option);
-    }
 
-    private static enum After implements AtAfterCondition {
-        INSTANCE() {
-            @Override
-            public boolean matches(BigInteger providedSequenceNumber, BigInteger shardSequenceNumber)
{
-                return providedSequenceNumber.compareTo(shardSequenceNumber) < 0;
-            }
-        }
-    }
 
-    private static enum At implements AtAfterCondition {
-        INSTANCE() {
-            @Override
-            public boolean matches(BigInteger providedSequenceNumber, BigInteger shardSequenceNumber)
{
-                return providedSequenceNumber.compareTo(shardSequenceNumber) <= 0;
-            }
-        }
-    }
+
 
     private static enum StartingSequenceNumberComparator implements Comparator<Shard>
{
         INSTANCE() {

http://git-wip-us.apache.org/repos/asf/camel/blob/8e05657d/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/AtAfterConditionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/AtAfterConditionTest.java
b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/AtAfterConditionTest.java
new file mode 100644
index 0000000..53aee40
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/AtAfterConditionTest.java
@@ -0,0 +1,67 @@
+/**
+ * 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.component.aws.ddbstream;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+
+@RunWith(Parameterized.class)
+public class AtAfterConditionTest {
+
+
+    private final AtAfterCondition condition;
+    private final int smaller;
+    private final int bigger;
+    private final boolean result;
+
+    public AtAfterConditionTest(AtAfterCondition condition, int smaller, int bigger, boolean
result) {
+        this.condition = condition;
+        this.smaller = smaller;
+        this.bigger = bigger;
+        this.result = result;
+    }
+
+    @Parameterized.Parameters
+    public static Collection<Object[]> parameters() {
+        List<Object[]> results = new ArrayList<>();
+
+        results.add(new Object[]{AtAfterCondition.Conditions.AFTER, 1, 5, true});
+        results.add(new Object[]{AtAfterCondition.Conditions.AT   , 1, 5, true});
+        results.add(new Object[]{AtAfterCondition.Conditions.AFTER, 1, 1, false});
+        results.add(new Object[]{AtAfterCondition.Conditions.AT   , 1, 1, true});
+        results.add(new Object[]{AtAfterCondition.Conditions.AFTER, 5, 1, false});
+        results.add(new Object[]{AtAfterCondition.Conditions.AT   , 5, 1, false});
+
+        return results;
+    }
+
+    @Test
+    public void test() throws Exception {
+        assertThat(condition.matches(BigInteger.valueOf(smaller), BigInteger.valueOf(bigger)),
is(result));
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/8e05657d/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
index 3ef6c70..0cbf9ca 100644
--- a/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/ddbstream/DdbStreamConsumerTest.java
@@ -1,3 +1,19 @@
+/**
+ * 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.component.aws.ddbstream;
 
 import java.util.ArrayList;
@@ -110,12 +126,15 @@ public class DdbStreamConsumerTest {
             @Override
             public GetRecordsResult answer(InvocationOnMock invocation) throws Throwable
{
                 final String shardIterator = ((GetRecordsRequest) invocation.getArguments()[0]).getShardIterator();
-                String nextShardIterator = shardIterators.get(shardIterator); // note that
HashMap returns null when there is no entry in the map. A null 'nextShardIterator' indicates
that the shard has finished and we should move onto the next shard.
+                // note that HashMap returns null when there is no entry in the map.
+                // A null 'nextShardIterator' indicates that the shard has finished
+                // and we should move onto the next shard.
+                String nextShardIterator = shardIterators.get(shardIterator);
                 Matcher m = Pattern.compile("shard_iterator_d_0*(\\d+)").matcher(shardIterator);
                 Collection<Record> ans = answers.get(shardIterator);
                 if (nextShardIterator == null && m.matches()) { // last shard iterates
forever.
                     Integer num = Integer.parseInt(m.group(1));
-                    nextShardIterator = "shard_iterator_d_" + pad(Integer.toString(num+1),
3);
+                    nextShardIterator = "shard_iterator_d_" + pad(Integer.toString(num +
1), 3);
                 }
                 if (null == ans) { // default to an empty list of records.
                     ans = createRecords();
@@ -227,6 +246,39 @@ public class DdbStreamConsumerTest {
         assertThat(getIteratorCaptor.getValue().getShardId(), is("c"));
     }
 
+    @Test
+    public void atSeqNumber35GivesFirstRecordWithSeq35() throws Exception {
+        endpoint.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER);
+        endpoint.setSequenceNumberProvider(new StaticSequenceNumberProvider("35"));
+        undertest = new DdbStreamConsumer(endpoint, processor);
+
+        for (int i = 0; i < 10; ++i) { // poll lots.
+            undertest.poll();
+        }
+
+        ArgumentCaptor<Exchange> exchangeCaptor = ArgumentCaptor.forClass(Exchange.class);
+        verify(processor, times(2)).process(exchangeCaptor.capture(), any(AsyncCallback.class));
+
+        assertThat(exchangeCaptor.getAllValues().get(0).getIn().getBody(Record.class).getDynamodb().getSequenceNumber(),
is("35"));
+        assertThat(exchangeCaptor.getAllValues().get(1).getIn().getBody(Record.class).getDynamodb().getSequenceNumber(),
is("40"));
+    }
+
+    @Test
+    public void afterSeqNumber35GivesFirstRecordWithSeq40() throws Exception {
+        endpoint.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
+        endpoint.setSequenceNumberProvider(new StaticSequenceNumberProvider("35"));
+        undertest = new DdbStreamConsumer(endpoint, processor);
+
+        for (int i = 0; i < 10; ++i) { // poll lots.
+            undertest.poll();
+        }
+
+        ArgumentCaptor<Exchange> exchangeCaptor = ArgumentCaptor.forClass(Exchange.class);
+        verify(processor, times(1)).process(exchangeCaptor.capture(), any(AsyncCallback.class));
+
+        assertThat(exchangeCaptor.getAllValues().get(0).getIn().getBody(Record.class).getDynamodb().getSequenceNumber(),
is("40"));
+    }
+
     private static Collection<Record> createRecords(String... sequenceNumbers) {
         List<Record> results = new ArrayList<>();
 


Mime
View raw message