You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by cm...@apache.org on 2017/03/26 10:50:56 UTC
camel git commit: added integration test for Amazon Kinesis Streams
integration
Repository: camel
Updated Branches:
refs/heads/master 833a1a301 -> fce2a5ab5
added integration test for Amazon Kinesis Streams integration
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fce2a5ab
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fce2a5ab
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fce2a5ab
Branch: refs/heads/master
Commit: fce2a5ab51ec7371fa77a354856ae19b3f5862fb
Parents: 833a1a3
Author: cmueller <ch...@gmail.com>
Authored: Sun Mar 26 12:50:16 2017 +0200
Committer: cmueller <ch...@gmail.com>
Committed: Sun Mar 26 12:50:16 2017 +0200
----------------------------------------------------------------------
.../KinesisComponentIntegrationTest.java | 103 +++++++++++++++++++
1 file changed, 103 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/fce2a5ab/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/integration/KinesisComponentIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/integration/KinesisComponentIntegrationTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/integration/KinesisComponentIntegrationTest.java
new file mode 100644
index 0000000..5764c59
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/kinesis/integration/KinesisComponentIntegrationTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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.integration;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.aws.kinesis.KinesisConstants;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
+import com.amazonaws.services.kinesis.model.Record;
+
+@Ignore("Must be manually tested.")
+public class KinesisComponentIntegrationTest extends CamelTestSupport {
+
+ @EndpointInject(uri = "direct:start")
+ private ProducerTemplate template;
+
+ @EndpointInject(uri = "mock:result")
+ private MockEndpoint result;
+
+ @Test
+ public void send() throws Exception {
+ result.expectedMessageCount(2);
+
+ template.send("direct:start", ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(KinesisConstants.PARTITION_KEY, "partition-1");
+ exchange.getIn().setBody("Kinesis Event 1.");
+ }
+ });
+
+ template.send("direct:start", ExchangePattern.InOut, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(KinesisConstants.PARTITION_KEY, "partition-1");
+ exchange.getIn().setBody("Kinesis Event 2.");
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+
+ assertResultExchange(result.getExchanges().get(0), "Kinesis Event 1.", "partition-1");
+ assertResultExchange(result.getExchanges().get(1), "Kinesis Event 2.", "partition-1");
+ }
+
+ private void assertResultExchange(Exchange resultExchange, String data, String partition) {
+ assertIsInstanceOf(Record.class, resultExchange.getIn().getBody());
+ Record record = resultExchange.getIn().getBody(Record.class);
+ assertEquals(data, new String(record.getData().array()));
+ assertEquals(partition, resultExchange.getIn().getHeader(KinesisConstants.PARTITION_KEY));
+ assertNotNull(resultExchange.getIn().getHeader(KinesisConstants.APPROX_ARRIVAL_TIME));
+ assertNotNull(resultExchange.getIn().getHeader(KinesisConstants.SEQUENCE_NUMBER));
+ }
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry registry = super.createRegistry();
+
+ AmazonKinesis client = AmazonKinesisClientBuilder.standard().withRegion(Regions.EU_CENTRAL_1).build();
+ registry.bind("amazonKinesisClient", client);
+
+ return registry;
+ }
+
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ String kinesisEndpointUri = "aws-kinesis://etl?amazonKinesisClient=#amazonKinesisClient";
+
+ from("direct:start")
+ .to(kinesisEndpointUri);
+
+ from(kinesisEndpointUri)
+ .to("mock:result");
+ }
+ };
+ }
+}
\ No newline at end of file