camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/3] camel git commit: Added an endpoint for consuming Amazon Kinesis Streams.
Date Sat, 05 Dec 2015 09:41:23 GMT
Repository: camel
Updated Branches:
  refs/heads/master 19e63dbaf -> e78a02960


Added an endpoint for consuming Amazon Kinesis Streams.


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

Branch: refs/heads/master
Commit: 72370df11c9768b697a930e22c417f1af95dd919
Parents: 19e63db
Author: Candle <candle@candle.me.uk>
Authored: Wed Dec 2 16:12:48 2015 +0000
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Sat Dec 5 10:06:17 2015 +0100

----------------------------------------------------------------------
 .../component/aws/kinesis/KinesisComponent.java |  44 +++++
 .../component/aws/kinesis/KinesisConstants.java |  29 ++++
 .../component/aws/kinesis/KinesisConsumer.java  | 126 +++++++++++++++
 .../component/aws/kinesis/KinesisEndpoint.java  | 123 ++++++++++++++
 .../aws/kinesis/RecordStringConverter.java      |  43 +++++
 .../services/org/apache/camel/TypeConverter     |  18 +++
 .../org/apache/camel/component/aws-kinesis      |  18 +++
 .../aws/kinesis/KinesisConsumerTest.java        | 162 +++++++++++++++++++
 .../aws/kinesis/KinesisEndpointTest.java        |  71 ++++++++
 .../aws/kinesis/RecordStringConverterTest.java  |  38 +++++
 10 files changed, 672 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisComponent.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisComponent.java
new file mode 100644
index 0000000..d3f34ab
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisComponent.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.kinesis;
+
+import java.util.Map;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.UriEndpointComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class KinesisComponent extends UriEndpointComponent {
+    private static final Logger LOG = LoggerFactory.getLogger(KinesisComponent.class);
+
+    public KinesisComponent() {
+        super(KinesisEndpoint.class);
+    }
+
+    public KinesisComponent(CamelContext context) {
+        super(context, KinesisEndpoint.class);
+    }
+
+    @Override
+    protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object>
parameters) throws Exception {
+        KinesisEndpoint endpoint = new KinesisEndpoint(uri, remaining, this);
+
+        LOG.debug("Created endpoint: {}", endpoint.toString());
+        return endpoint;
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConstants.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConstants.java
new file mode 100644
index 0000000..b028123
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConstants.java
@@ -0,0 +1,29 @@
+/**
+ * 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.kinesis;
+
+public interface KinesisConstants {
+
+    public static final String SEQUENCE_NUMBER = "KinesisSequenceNumber";
+    public static final String APPROX_ARRIVAL_TIME = "KinesisApproximateArrivalTimestamp";
+    public static final String PARTITION_KEY = "KinesisPartitionKey";
+    /**
+     * in a Kinesis Record object, the shard ID is obtained from the getPartitionKey method.
+     */
+    public static final String SHARD_ID = "KinesisPartitionKey";
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConsumer.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConsumer.java
new file mode 100644
index 0000000..b301f38
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisConsumer.java
@@ -0,0 +1,126 @@
+/**
+ * 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.kinesis;
+
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.Record;
+import java.util.ArrayDeque;
+import java.util.List;
+import java.util.Queue;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.ScheduledBatchPollingConsumer;
+import org.apache.camel.util.CastUtils;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class KinesisConsumer extends ScheduledBatchPollingConsumer {
+    private static final Logger LOG = LoggerFactory.getLogger(KinesisConsumer.class);
+
+    private String currentShardIterator = null;
+
+    public KinesisConsumer(KinesisEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+    }
+
+    /*
+     * Returns the number of messages polled.
+     */
+    @Override
+    protected int poll() throws Exception {
+        GetRecordsRequest req = new GetRecordsRequest()
+                .withShardIterator(getShardItertor())
+                .withLimit(getEndpoint().getMaxResultsPerRequest())
+                ;
+        GetRecordsResult result = getClient().getRecords(req);
+
+        Queue<Exchange> exchanges = createExchanges(result.getRecords());
+        int processedExchangeCount = processBatch(CastUtils.cast(exchanges));
+
+        // May cache the last successful sequence number, and pass it to the
+        // getRecords request. That way, on the next poll, we start from where
+        // we left off, however, I don't know what happens to subsiquent
+        // exchanges when an earlier echange fails.
+
+        currentShardIterator = result.getNextShardIterator();
+
+        return processedExchangeCount;
+    }
+
+    @Override
+    public int processBatch(Queue<Object> exchanges) throws Exception {
+        int processedExchanges = 0;
+        while (!exchanges.isEmpty()) {
+            final Exchange exchange = ObjectHelper.cast(Exchange.class, exchanges.poll());
+
+            LOG.trace("Processing exchange [{}] started.", exchange);
+            getAsyncProcessor().process(exchange, new AsyncCallback() {
+                @Override
+                public void done(boolean doneSync) {
+                    LOG.trace("Processing exchange [{}] done.", exchange);
+                }
+            });
+            processedExchanges++;
+        }
+        return processedExchanges;
+    }
+
+    private AmazonKinesis getClient() {
+        return getEndpoint().getClient();
+    }
+
+    @Override
+    public KinesisEndpoint getEndpoint() {
+        return (KinesisEndpoint) super.getEndpoint();
+    }
+
+    private String getShardItertor() {
+        // either return a cached one or get a new one via a GetShardIterator request.
+        if (currentShardIterator == null) {
+            DescribeStreamRequest req1 = new DescribeStreamRequest()
+                    .withStreamName(getEndpoint().getStreamName())
+                    ;
+            DescribeStreamResult res1 = getClient().describeStream(req1);
+
+            GetShardIteratorRequest req = new GetShardIteratorRequest()
+                    .withStreamName(getEndpoint().getStreamName())
+                    .withShardId(res1.getStreamDescription().getShards().get(0).getShardId())
// XXX only uses the first shard
+                    .withShardIteratorType(getEndpoint().getIteratorType())
+                    ;
+            GetShardIteratorResult result = getClient().getShardIterator(req);
+            currentShardIterator = result.getShardIterator();
+        }
+        LOG.debug("Shard Iterator is: {}", currentShardIterator);
+        return currentShardIterator;
+    }
+
+    private Queue<Exchange> createExchanges(List<Record> records) {
+        Queue<Exchange> exchanges = new ArrayDeque<>();
+        for (Record record : records) {
+            exchanges.add(getEndpoint().createExchange(record));
+        }
+        return exchanges;
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisEndpoint.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisEndpoint.java
new file mode 100644
index 0000000..b4c7597
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/KinesisEndpoint.java
@@ -0,0 +1,123 @@
+/**
+ * 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.kinesis;
+
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.ScheduledPollEndpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+
+@UriEndpoint(scheme = "aws-kinesis", title = "AWS Kinesis", syntax = "aws-kinesis:streamName",
consumerClass = KinesisConsumer.class, label = "cloud,messaging")
+public class KinesisEndpoint extends ScheduledPollEndpoint {
+
+    @UriPath(label = "consumer,producer", description = "Name of the stream")
+    @Metadata(required = "true")
+    private String streamName;
+
+    // For now, always assume that we've been supplied a client in the Camel registry.
+    @UriParam(label = "consumer", description = "Amazon Kinesis client to use for all requests
for this endpoint")
+    @Metadata(required = "true")
+    private AmazonKinesis amazonKinesisClient;
+
+    @UriParam(label = "consumer", description = "Maximum number of records that will be fetched
in each poll")
+    private int maxResultsPerRequest = 1;
+
+    @UriParam(label = "consumer", description = "Defines where in the Kinesis stream to start
getting records")
+    private ShardIteratorType iteratorType = ShardIteratorType.TRIM_HORIZON;
+
+    public KinesisEndpoint(String uri, String streamName, KinesisComponent component) {
+        super(uri, component);
+        this.streamName = streamName;
+    }
+
+    @Override
+    public Producer createProducer() throws Exception {
+        throw new UnsupportedOperationException("Not supported yet.");
+    }
+
+    @Override
+    public Consumer createConsumer(Processor processor) throws Exception {
+        return new KinesisConsumer(this, processor);
+    }
+
+    Exchange createExchange(Record record) {
+        Exchange ex = super.createExchange();
+        ex.getIn().setBody(record, Record.class);
+        ex.setProperty(KinesisConstants.APPROX_ARRIVAL_TIME, record.getApproximateArrivalTimestamp());
+        ex.setProperty(KinesisConstants.PARTITION_KEY, record.getPartitionKey());
+        ex.setProperty(KinesisConstants.SEQUENCE_NUMBER, record.getSequenceNumber());
+
+        return ex;
+    }
+
+    @Override
+    public boolean isSingleton() {
+        // probably right.
+        return true;
+    }
+
+    AmazonKinesis getClient() {
+        return amazonKinesisClient;
+    }
+
+    // required for injection.
+    public AmazonKinesis getAmazonKinesisClient() {
+        return amazonKinesisClient;
+    }
+
+    public void setAmazonKinesisClient(AmazonKinesis amazonKinesisClient) {
+        this.amazonKinesisClient = amazonKinesisClient;
+    }
+
+    public int getMaxResultsPerRequest() {
+        return maxResultsPerRequest;
+    }
+
+    public void setMaxResultsPerRequest(int maxResultsPerRequest) {
+        this.maxResultsPerRequest = maxResultsPerRequest;
+    }
+
+    public String getStreamName() {
+        return streamName;
+    }
+
+    public void setStreamName(String streamName) {
+        this.streamName = streamName;
+    }
+
+    public ShardIteratorType getIteratorType() {
+        return iteratorType;
+    }
+
+    public void setIteratorType(ShardIteratorType iteratorType) {
+        this.iteratorType = iteratorType;
+    }
+
+    @Override
+    public String toString() {
+        return "KinesisEndpoint{" + "amazonKinesisClient=[redacted], maxResultsPerRequest="
+ maxResultsPerRequest + ", iteratorType=" + iteratorType + ", streamName=" + streamName +
'}';
+    }
+    
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/RecordStringConverter.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/RecordStringConverter.java
b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/RecordStringConverter.java
new file mode 100644
index 0000000..bda8983
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/kinesis/RecordStringConverter.java
@@ -0,0 +1,43 @@
+/**
+ * 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.kinesis;
+
+import com.amazonaws.services.kinesis.model.Record;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.camel.Converter;
+
+@Converter
+public class RecordStringConverter {
+
+    @Converter
+    public static String toString(Record record) {
+        List<Byte> bytes = new ArrayList<>();
+        ByteBuffer buf = record.getData().asReadOnlyBuffer();
+        while (buf.hasRemaining()) {
+            bytes.add(buf.get());
+        }
+        byte[] a = new byte[bytes.size()];
+        for (int i = 0; i < bytes.size(); ++i) {
+            a[i] = bytes.get(i);
+        }
+        return new String(a, Charset.forName("UTF-8"));
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/TypeConverter
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/TypeConverter
b/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/TypeConverter
new file mode 100644
index 0000000..28707d6
--- /dev/null
+++ b/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/TypeConverter
@@ -0,0 +1,18 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+org.apache.camel.component.aws.kinesis

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis
b/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis
new file mode 100644
index 0000000..2329cbf
--- /dev/null
+++ b/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis
@@ -0,0 +1,18 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+class=org.apache.camel.component.aws.kinesis.KinesisComponent

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisConsumerTest.java
b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisConsumerTest.java
new file mode 100644
index 0000000..db0df68
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisConsumerTest.java
@@ -0,0 +1,162 @@
+/**
+ * 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.kinesis;
+
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import com.amazonaws.services.kinesis.model.StreamDescription;
+import java.util.Date;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import static org.hamcrest.CoreMatchers.is;
+import org.junit.Test;
+import static org.junit.Assert.*;
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import static org.mockito.Mockito.*;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisConsumerTest {
+
+    @Mock private AmazonKinesis kinesisClient;
+    @Mock private AsyncProcessor processor;
+
+    private final CamelContext context = new DefaultCamelContext();
+    private final KinesisComponent component = new KinesisComponent(context);
+
+    private KinesisConsumer undertest;
+
+    @Before
+    public void setup() throws Exception {
+        KinesisEndpoint endpoint = new KinesisEndpoint(null, "streamName", component);
+        endpoint.setAmazonKinesisClient(kinesisClient);
+        endpoint.setIteratorType(ShardIteratorType.LATEST);
+        undertest = new KinesisConsumer(endpoint, processor);
+
+        when(kinesisClient.getRecords(any(GetRecordsRequest.class)))
+                .thenReturn(new GetRecordsResult()
+                        .withNextShardIterator("nextShardIterator")
+                );
+        when(kinesisClient.describeStream(any(DescribeStreamRequest.class)))
+                .thenReturn(new DescribeStreamResult()
+                        .withStreamDescription(new StreamDescription()
+                                .withShards(new Shard().withShardId("shardId"))
+                        )
+                );
+        when(kinesisClient.getShardIterator(any(GetShardIteratorRequest.class)))
+                .thenReturn(new GetShardIteratorResult()
+                        .withShardIterator("shardIterator")
+                );
+    }
+
+    @Test
+    public void itObtainsAShardIteratorOnFirstPoll() throws Exception {
+        undertest.poll();
+
+        final ArgumentCaptor<DescribeStreamRequest> describeStreamReqCap = ArgumentCaptor.forClass(DescribeStreamRequest.class);
+        final ArgumentCaptor<GetShardIteratorRequest> getShardIteratorReqCap = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
+
+        verify(kinesisClient).describeStream(describeStreamReqCap.capture());
+        assertThat(describeStreamReqCap.getValue().getStreamName(), is("streamName"));
+
+        verify(kinesisClient).getShardIterator(getShardIteratorReqCap.capture());
+        assertThat(getShardIteratorReqCap.getValue().getStreamName(), is("streamName"));
+        assertThat(getShardIteratorReqCap.getValue().getShardId(), is("shardId"));
+        assertThat(getShardIteratorReqCap.getValue().getShardIteratorType(), is("LATEST"));
+    }
+
+    @Test
+    public void itUsesTheShardIteratorOnPolls() throws Exception {
+        undertest.poll();
+
+        final ArgumentCaptor<GetRecordsRequest> getRecordsReqCap = ArgumentCaptor.forClass(GetRecordsRequest.class);
+        verify(kinesisClient).getRecords(getRecordsReqCap.capture());
+
+        assertThat(getRecordsReqCap.getValue().getShardIterator(), is("shardIterator"));
+    }
+
+
+    @Test
+    public void itUsesTheShardIteratorOnSubsiquentPolls() throws Exception {
+        undertest.poll();
+        undertest.poll();
+
+        final ArgumentCaptor<GetRecordsRequest> getRecordsReqCap = ArgumentCaptor.forClass(GetRecordsRequest.class);
+
+        verify(kinesisClient, times(1)).describeStream(any(DescribeStreamRequest.class));
+        verify(kinesisClient, times(1)).getShardIterator(any(GetShardIteratorRequest.class));
+        verify(kinesisClient, times(2)).getRecords(getRecordsReqCap.capture());
+        assertThat(getRecordsReqCap.getAllValues().get(0).getShardIterator(), is("shardIterator"));
+        assertThat(getRecordsReqCap.getAllValues().get(1).getShardIterator(), is("nextShardIterator"));
+    }
+
+    @Test
+    public void recordsAreSentToTheProcessor() throws Exception {
+        when(kinesisClient.getRecords(any(GetRecordsRequest.class)))
+                .thenReturn(new GetRecordsResult()
+                        .withNextShardIterator("nextShardIterator")
+                        .withRecords(new Record().withSequenceNumber("1"), new Record().withSequenceNumber("2"))
+                );
+
+        int messageCount = undertest.poll();
+
+        assertThat(messageCount, is(2));
+        final 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).getSequenceNumber(),
is("1"));
+        assertThat(exchangeCaptor.getAllValues().get(1).getIn().getBody(Record.class).getSequenceNumber(),
is("2"));
+    }
+
+    @Test
+    public void exchangePropertiesAreSet() throws Exception {
+
+        when(kinesisClient.getRecords(any(GetRecordsRequest.class)))
+                .thenReturn(new GetRecordsResult()
+                        .withNextShardIterator("nextShardIterator")
+                        .withRecords(new Record()
+                                .withSequenceNumber("1")
+                                .withApproximateArrivalTimestamp(new Date(42))
+                                .withPartitionKey("shardId")
+                        )
+                );
+
+        undertest.poll();
+
+        final ArgumentCaptor<Exchange> exchangeCaptor = ArgumentCaptor.forClass(Exchange.class);
+
+        verify(processor).process(exchangeCaptor.capture(), any(AsyncCallback.class));
+        assertThat(exchangeCaptor.getValue().getProperty(KinesisConstants.APPROX_ARRIVAL_TIME,
long.class), is(42L));
+        assertThat(exchangeCaptor.getValue().getProperty(KinesisConstants.PARTITION_KEY,
String.class), is("shardId"));
+        assertThat(exchangeCaptor.getValue().getProperty(KinesisConstants.SEQUENCE_NUMBER,
String.class), is("1"));
+        assertThat(exchangeCaptor.getValue().getProperty(KinesisConstants.SHARD_ID, String.class),
is("shardId"));
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisEndpointTest.java
b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisEndpointTest.java
new file mode 100644
index 0000000..50653e3
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/KinesisEndpointTest.java
@@ -0,0 +1,71 @@
+/**
+ * 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.kinesis;
+
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.SimpleRegistry;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisEndpointTest {
+
+    @Mock private AmazonKinesis amazonKinesisClient;
+
+    private CamelContext camelContext;
+
+    @Before
+    public void setup() throws Exception {
+        SimpleRegistry registry = new SimpleRegistry();
+        registry.put("kinesisClient", amazonKinesisClient);
+        camelContext = new DefaultCamelContext(registry);
+    }
+
+    @Test
+    public void allTheEndpointParams() throws Exception {
+        KinesisEndpoint endpoint = (KinesisEndpoint)camelContext.getEndpoint("aws-kinesis://some_stream_name"
+                + "?amazonKinesisClient=#kinesisClient"
+                + "&maxResultsPerRequest=101"
+                + "&iteratorType=latest"
+                );
+
+        assertThat(endpoint.getClient(), is(amazonKinesisClient));
+        assertThat(endpoint.getStreamName(), is("some_stream_name"));
+        assertThat(endpoint.getIteratorType(), is(ShardIteratorType.LATEST));
+        assertThat(endpoint.getMaxResultsPerRequest(), is(101));
+    }
+
+    @Test
+    public void onlyRequiredEndpointParams() throws Exception {
+        KinesisEndpoint endpoint = (KinesisEndpoint)camelContext.getEndpoint("aws-kinesis://some_stream_name"
+                + "?amazonKinesisClient=#kinesisClient"
+                );
+
+        assertThat(endpoint.getClient(), is(amazonKinesisClient));
+        assertThat(endpoint.getStreamName(), is("some_stream_name"));
+        assertThat(endpoint.getIteratorType(), is(ShardIteratorType.TRIM_HORIZON));
+        assertThat(endpoint.getMaxResultsPerRequest(), is(1));
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/72370df1/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/RecordStringConverterTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/RecordStringConverterTest.java
b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/RecordStringConverterTest.java
new file mode 100644
index 0000000..48f8edb
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/RecordStringConverterTest.java
@@ -0,0 +1,38 @@
+/**
+ * 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.kinesis;
+
+import com.amazonaws.services.kinesis.model.Record;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import static org.hamcrest.CoreMatchers.is;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class RecordStringConverterTest {
+
+    @Test
+    public void convertRecordToString() throws Exception {
+        Record record = new Record()
+                .withSequenceNumber("1")
+                .withData(ByteBuffer.wrap("this is a String".getBytes(Charset.forName("UTF-8"))))
+                ;
+
+        String result = RecordStringConverter.toString(record);
+        assertThat(result, is("this is a String"));
+    }
+}
\ No newline at end of file


Mime
View raw message