You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2017/03/08 07:48:19 UTC
[1/2] camel git commit: CAMEL-10952 Initial work adding Kinesis
Firehose
Repository: camel
Updated Branches:
refs/heads/master 8d33bc188 -> 98ba70378
CAMEL-10952 Initial work adding Kinesis Firehose
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f3f02f01
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f3f02f01
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f3f02f01
Branch: refs/heads/master
Commit: f3f02f018540039ced717e9b84f596472b15f630
Parents: 8d33bc1
Author: Tony Tiger <th...@gmail.com>
Authored: Mon Mar 6 16:34:03 2017 +0000
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Wed Mar 8 08:40:51 2017 +0100
----------------------------------------------------------------------
components/camel-aws/pom.xml | 1 +
.../docs/aws-kinesis-firehose-component.adoc | 146 +++++++++++++++++++
.../aws/firehose/KinesisFirehoseComponent.java | 41 ++++++
.../aws/firehose/KinesisFirehoseConstants.java | 22 +++
.../aws/firehose/KinesisFirehoseEndpoint.java | 72 +++++++++
.../aws/firehose/KinesisFirehoseProducer.java | 61 ++++++++
.../apache/camel/component/aws-kinesis-firehose | 18 +++
.../firehose/KinesisFirehoseEndpointTest.java | 56 +++++++
.../firehose/KinesisFirehoseProducerTest.java | 82 +++++++++++
...nesisFirehoseComponentAutoConfiguration.java | 80 ++++++++++
...dditional-spring-configuration-metadata.json | 6 +
.../main/resources/META-INF/spring.factories | 4 +-
12 files changed, 588 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-aws/pom.xml b/components/camel-aws/pom.xml
index 43da218..0fe9938 100644
--- a/components/camel-aws/pom.xml
+++ b/components/camel-aws/pom.xml
@@ -37,6 +37,7 @@
org.apache.camel.spi.ComponentResolver;component=aws-ddb,
org.apache.camel.spi.ComponentResolver;component=aws-ec2,
org.apache.camel.spi.ComponentResolver;component=aws-kinesis,
+ org.apache.camel.spi.ComponentResolver;component=aws-kinesis-firehose,
org.apache.camel.spi.ComponentResolver;component=aws-s3,
org.apache.camel.spi.ComponentResolver;component=aws-sdb,
org.apache.camel.spi.ComponentResolver;component=aws-ses,
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc b/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
new file mode 100644
index 0000000..b691073
--- /dev/null
+++ b/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
@@ -0,0 +1,146 @@
+## AWS Kinesis Firehose Component
+
+*Available as of Camel version 2.19*
+
+The Kinesis Firehose component supports sending messages to Amazon Kinesis Firehose service.
+
+Prerequisites
+
+You must have a valid Amazon Web Services developer account, and be
+signed up to use Amazon Kinesis Firehose. More information are available
+at�https://aws.amazon.com/kinesis/firehose/[AWS Kinesis Firehose]
+
+### URI Format
+
+[source,java]
+-----------------------------------
+aws-kinesis-firehose://delivery-stream-name[?options]
+-----------------------------------
+
+The stream needs to be created prior to it being used. +
+ You can append query options to the URI in the following format,
+?options=value&option2=value&...
+
+### URI Options
+
+
+// component options: START
+The AWS Kinesis Firehose component has no options.
+// component options: END
+
+
+
+
+
+
+
+// endpoint options: START
+The AWS Kinesis Firehose component is configured using the URI syntax with the following path and query parameters:
+
+ aws-kinesis-firehose:streamName
+
+#### 1 path parameters:
+
+[width="100%",cols="2,1,1m,6",options="header"]
+|=======================================================================
+| Name | Default | Java Type | Description
+| streamName | | String | *Required* Name of the delivery stream
+|=======================================================================
+
+#### 5 query parameters:
+
+[width="100%",cols="2,1,1m,1m,5",options="header"]
+|=======================================================================
+| Name | Group | Default | Java Type | Description
+| amazonKinesisFirehoseClient | common | | AmazonKinesisFirehose | *Required* Amazon Kinesis Firehose client to use for all requests for this endpoint
+| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
+|=======================================================================
+// endpoint options: END
+
+
+
+
+
+
+Required Kinesis Firehose component options
+
+You have to provide the�amazonKinesisClient in the
+link:registry.html[Registry] with proxies and relevant credentials
+configured.
+
+
+### Usage
+
+#### Amazon Kinesis Firehose configuration
+
+You will need to create an instance of AmazonKinesisClient and
+bind it to the registry
+
+[source,java]
+--------------------------------------------------------------------------------------------------------------------
+ClientConfiguration clientConfiguration = new ClientConfiguration();
+clientConfiguration.setProxyHost("http://myProxyHost");
+clientConfiguration.setProxyPort(8080);
+
+Region region = Region.getRegion(Regions.fromName(region));
+region.createClient(AmazonKinesisClient.class, null, clientConfiguration);
+// the 'null' here is the AWSCredentialsProvider which defaults to an instance of DefaultAWSCredentialsProviderChain
+
+registry.bind("kinesisFirehoseClient", client);
+--------------------------------------------------------------------------------------------------------------------
+
+You then have to reference the AmazonKinesisFirehoseClient in the `amazonKinesisFirehoseClient` URI option.
+
+[source,java]
+--------------------------------------------------------------------------------------------------------------------
+from("aws-kinesis-firehose://mykinesisdeliverystream?amazonKinesisFirehoseClient=#kinesisClient")
+ .to("log:out?showAll=true");
+--------------------------------------------------------------------------------------------------------------------
+
+#### Providing AWS Credentials
+
+It is recommended that the credentials are obtained by using the
+http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html[DefaultAWSCredentialsProviderChain]
+that is the default when creating a new ClientConfiguration instance,
+however, a
+different�http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html[AWSCredentialsProvider]
+can be specified when calling createClient(...).
+
+#### Message headers set by the Kinesis producer on successful storage of a Record
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Header |Type |Description
+
+|`CamelAwsKinesisFirehoseRecordId` |`String` |The record ID, as defined in
+http://docs.aws.amazon.com/firehose/latest/APIReference/API_PutRecord.html#API_PutRecord_ResponseSyntax[Response Syntax]
+
+
+|=======================================================================
+
+### Dependencies
+
+Maven users will need to add the following dependency to their pom.xml.
+
+*pom.xml*
+
+[source,xml]
+---------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-aws</artifactId>
+ <version>${camel-version}</version>
+</dependency>
+---------------------------------------
+
+where `${camel-version`} must be replaced by the actual version of Camel
+(2.19 or higher).
+
+### See Also
+
+* link:configuring-camel.html[Configuring Camel]
+* link:component.html[Component]
+* link:endpoint.html[Endpoint]
+* link:getting-started.html[Getting Started]
+
+* link:aws.html[AWS Component]
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseComponent.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseComponent.java
new file mode 100644
index 0000000..0df1504
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseComponent.java
@@ -0,0 +1,41 @@
+/**
+ * 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.firehose;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultComponent;
+
+public class KinesisFirehoseComponent extends DefaultComponent {
+
+ public KinesisFirehoseComponent() {
+
+ }
+
+ public KinesisFirehoseComponent(CamelContext context) {
+ super(context);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ KinesisFirehoseEndpoint endpoint = new KinesisFirehoseEndpoint(uri, remaining, this);
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseConstants.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseConstants.java
new file mode 100644
index 0000000..b85822e
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseConstants.java
@@ -0,0 +1,22 @@
+/**
+ * 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.firehose;
+
+public interface KinesisFirehoseConstants {
+
+ String RECORD_ID = "CamelAwsKinesisFirehoseRecordId";
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpoint.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpoint.java
new file mode 100644
index 0000000..b5ee50c
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpoint.java
@@ -0,0 +1,72 @@
+/**
+ * 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.firehose;
+
+import com.amazonaws.services.kinesisfirehose.AmazonKinesisFirehose;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.aws.kinesis.KinesisConsumer;
+import org.apache.camel.impl.DefaultEndpoint;
+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(firstVersion = "2.19.0", scheme = "aws-kinesis-firehose", title = "AWS Kinesis Firehose", syntax = "aws-kinesis-firehose:streamName", producerOnly = true,
+ consumerClass = KinesisConsumer.class, label = "cloud,messaging")
+public class KinesisFirehoseEndpoint extends DefaultEndpoint {
+
+ @UriPath(description = "Name of the stream")
+ @Metadata(required = "true")
+ private String streamName;
+ @UriParam(description = "Amazon Kinesis Firehose client to use for all requests for this endpoint")
+ @Metadata(required = "true")
+ private AmazonKinesisFirehose amazonKinesisFirehoseClient;
+
+ public KinesisFirehoseEndpoint(String uri, String streamName, KinesisFirehoseComponent component) {
+ super(uri, component);
+ this.streamName = streamName;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new KinesisFirehoseProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("You cannot consume messages from this endpoint");
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public void setAmazonKinesisFirehoseClient(AmazonKinesisFirehose client) {
+ this.amazonKinesisFirehoseClient = client;
+ }
+
+ public AmazonKinesisFirehose getClient() {
+ return amazonKinesisFirehoseClient;
+ }
+
+ public String getStreamName() {
+ return streamName;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducer.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducer.java
new file mode 100644
index 0000000..6eee663
--- /dev/null
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducer.java
@@ -0,0 +1,61 @@
+/**
+ * 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.firehose;
+
+import java.nio.ByteBuffer;
+
+import com.amazonaws.services.kinesisfirehose.model.PutRecordRequest;
+import com.amazonaws.services.kinesisfirehose.model.PutRecordResult;
+import com.amazonaws.services.kinesisfirehose.model.Record;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.impl.DefaultProducer;
+
+import static org.apache.camel.component.aws.common.AwsExchangeUtil.getMessageForResponse;
+
+public class KinesisFirehoseProducer extends DefaultProducer {
+
+ public KinesisFirehoseProducer(KinesisFirehoseEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public KinesisFirehoseEndpoint getEndpoint() {
+ return (KinesisFirehoseEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ PutRecordRequest request = createRequest(exchange);
+ log.trace("Sending request [{}] from exchange [{}]...", request, exchange);
+ PutRecordResult putRecordResult = getEndpoint().getClient().putRecord(request);
+ log.trace("Received result [{}]", putRecordResult);
+ Message message = getMessageForResponse(exchange);
+ message.setHeader(KinesisFirehoseConstants.RECORD_ID, putRecordResult.getRecordId());
+ }
+
+ private PutRecordRequest createRequest(Exchange exchange) {
+ ByteBuffer body = exchange.getIn().getBody(ByteBuffer.class);
+ Record record = new Record();
+ record.setData(body);
+
+ PutRecordRequest putRecordRequest = new PutRecordRequest();
+ putRecordRequest.setDeliveryStreamName(getEndpoint().getEndpointKey());
+ putRecordRequest.setRecord(record);
+ return putRecordRequest;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis-firehose
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis-firehose b/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis-firehose
new file mode 100644
index 0000000..2f91974
--- /dev/null
+++ b/components/camel-aws/src/main/resources/META-INF/services/org/apache/camel/component/aws-kinesis-firehose
@@ -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.firehose.KinesisFirehoseComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpointTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpointTest.java
new file mode 100644
index 0000000..2a0ef0f
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseEndpointTest.java
@@ -0,0 +1,56 @@
+/**
+ * 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.firehose;
+
+import com.amazonaws.services.kinesisfirehose.AmazonKinesisFirehose;
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.SimpleRegistry;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisFirehoseEndpointTest {
+
+ @Mock
+ private AmazonKinesisFirehose amazonKinesisFirehoseClient;
+
+ private CamelContext camelContext;
+
+ @Before
+ public void setup() throws Exception {
+ SimpleRegistry registry = new SimpleRegistry();
+ registry.put("firehoseClient", amazonKinesisFirehoseClient);
+ camelContext = new DefaultCamelContext(registry);
+ }
+
+ @Test
+ public void allEndpointParams() throws Exception {
+ KinesisFirehoseEndpoint endpoint = (KinesisFirehoseEndpoint) camelContext.getEndpoint("aws-kinesis-firehose://some_stream_name"
+ + "?amazonKinesisFirehoseClient=#firehoseClient"
+ );
+
+ assertThat(endpoint.getClient(), is(amazonKinesisFirehoseClient));
+ assertThat(endpoint.getStreamName(), is("some_stream_name"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducerTest.java b/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducerTest.java
new file mode 100644
index 0000000..2096714
--- /dev/null
+++ b/components/camel-aws/src/test/java/org/apache/camel/component/aws/firehose/KinesisFirehoseProducerTest.java
@@ -0,0 +1,82 @@
+/**
+ * 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.firehose;
+
+import java.nio.ByteBuffer;
+
+import com.amazonaws.services.kinesisfirehose.AmazonKinesisFirehose;
+import com.amazonaws.services.kinesisfirehose.model.PutRecordRequest;
+import com.amazonaws.services.kinesisfirehose.model.PutRecordResult;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Answers;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisFirehoseProducerTest {
+
+ private static final String STREAM_NAME = "streams";
+ private static final String RECORD_ID = "sample_record_id";
+ private static final String SAMPLE_RECORD_BODY = "SAMPLE";
+ private static final ByteBuffer SAMPLE_BUFFER = ByteBuffer.wrap(SAMPLE_RECORD_BODY.getBytes());
+
+ @Mock
+ private AmazonKinesisFirehose kinesisFirehoseClient;
+ @Mock
+ private KinesisFirehoseEndpoint kinesisFirehoseEndpoint;
+ @Mock
+ private Message inMessage;
+ @Mock
+ private Message outMessage;
+ @Mock
+ private PutRecordResult putRecordResult;
+ @Mock(answer = Answers.RETURNS_DEEP_STUBS)
+ private Exchange exchange;
+
+ private KinesisFirehoseProducer kinesisFirehoseProducer;
+ @Before
+ public void setup() throws Exception {
+ when(kinesisFirehoseEndpoint.getClient()).thenReturn(kinesisFirehoseClient);
+ when(kinesisFirehoseEndpoint.getEndpointUri()).thenReturn("aws-kinesis-firehose://etl");
+ when(kinesisFirehoseEndpoint.getStreamName()).thenReturn(STREAM_NAME);
+ when(exchange.getOut()).thenReturn(outMessage);
+ when(exchange.getIn()).thenReturn(inMessage);
+ when(exchange.getPattern()).thenReturn(ExchangePattern.InOut);
+
+ when(inMessage.getBody(ByteBuffer.class)).thenReturn(SAMPLE_BUFFER);
+
+ when(putRecordResult.getRecordId()).thenReturn(RECORD_ID);
+ when(kinesisFirehoseClient.putRecord(any(PutRecordRequest.class))).thenReturn(putRecordResult);
+ kinesisFirehoseProducer = new KinesisFirehoseProducer(kinesisFirehoseEndpoint);
+ }
+
+ @Test
+ public void shouldPutRecordIntoStreamWhenProcessingExchange() throws Exception {
+ kinesisFirehoseProducer.process(exchange);
+ verify(outMessage).setHeader(KinesisFirehoseConstants.RECORD_ID, RECORD_ID);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/platforms/spring-boot/components-starter/camel-aws-starter/src/main/java/org/apache/camel/component/aws/firehose/springboot/KinesisFirehoseComponentAutoConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-aws-starter/src/main/java/org/apache/camel/component/aws/firehose/springboot/KinesisFirehoseComponentAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-aws-starter/src/main/java/org/apache/camel/component/aws/firehose/springboot/KinesisFirehoseComponentAutoConfiguration.java
new file mode 100644
index 0000000..7f60733
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-aws-starter/src/main/java/org/apache/camel/component/aws/firehose/springboot/KinesisFirehoseComponentAutoConfiguration.java
@@ -0,0 +1,80 @@
+/**
+ * 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.firehose.springboot;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.aws.firehose.KinesisFirehoseComponent;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionMessage;
+import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
+import org.springframework.boot.bind.RelaxedPropertyResolver;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+/**
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Configuration
+@ConditionalOnBean(type = "org.apache.camel.spring.boot.CamelAutoConfiguration")
+@Conditional(KinesisFirehoseComponentAutoConfiguration.Condition.class)
+@AutoConfigureAfter(name = "org.apache.camel.spring.boot.CamelAutoConfiguration")
+public class KinesisFirehoseComponentAutoConfiguration {
+
+ @Lazy
+ @Bean(name = "aws-kinesis-firehose-component")
+ @ConditionalOnClass(CamelContext.class)
+ @ConditionalOnMissingBean(KinesisFirehoseComponent.class)
+ public KinesisFirehoseComponent configureKinesisFirehoseComponent(
+ CamelContext camelContext) throws Exception {
+ KinesisFirehoseComponent component = new KinesisFirehoseComponent();
+ component.setCamelContext(camelContext);
+ return component;
+ }
+
+ public static class Condition extends SpringBootCondition {
+ @Override
+ public ConditionOutcome getMatchOutcome(
+ ConditionContext conditionContext,
+ AnnotatedTypeMetadata annotatedTypeMetadata) {
+ boolean groupEnabled = isEnabled(conditionContext,
+ "camel.component.", true);
+ ConditionMessage.Builder message = ConditionMessage
+ .forCondition("camel.component.aws-kinesis-firehose");
+ if (isEnabled(conditionContext,
+ "camel.component.aws-kinesis-firehose.", groupEnabled)) {
+ return ConditionOutcome.match(message.because("enabled"));
+ }
+ return ConditionOutcome.noMatch(message.because("not enabled"));
+ }
+
+ private boolean isEnabled(
+ org.springframework.context.annotation.ConditionContext context,
+ java.lang.String prefix, boolean defaultValue) {
+ RelaxedPropertyResolver resolver = new RelaxedPropertyResolver(
+ context.getEnvironment(), prefix);
+ return resolver.getProperty("enabled", Boolean.class, defaultValue);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
index d043ada..ebf2a0a 100644
--- a/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
+++ b/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
@@ -65,6 +65,12 @@
"name": "camel.component.aws-ddb.enabled",
"description": "Enable aws-ddb component",
"type": "java.lang.Boolean"
+ },
+ {
+ "defaultValue": true,
+ "name": "camel.component.aws-kinesis-firehose.enabled",
+ "description": "Enable aws-kinesis-firehose component",
+ "type": "java.lang.Boolean"
}
]
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/f3f02f01/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/spring.factories
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/spring.factories b/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/spring.factories
index 729f3b3..49f3988 100644
--- a/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/spring.factories
+++ b/platforms/spring-boot/components-starter/camel-aws-starter/src/main/resources/META-INF/spring.factories
@@ -26,7 +26,9 @@ org.apache.camel.component.aws.ddbstream.springboot.DdbStreamComponentAutoConfig
org.apache.camel.component.aws.sqs.springboot.SqsComponentAutoConfiguration,\
org.apache.camel.component.aws.ec2.springboot.EC2ComponentAutoConfiguration,\
org.apache.camel.component.aws.cw.springboot.CwComponentAutoConfiguration,\
-org.apache.camel.component.aws.ddb.springboot.DdbComponentAutoConfiguration
+org.apache.camel.component.aws.ddb.springboot.DdbComponentAutoConfiguration,\
+org.apache.camel.component.aws.firehose.springboot.KinesisFirehoseComponentAutoConfiguration
+
[2/2] camel git commit: CAMEL-10952: Regen docs
Posted by ac...@apache.org.
CAMEL-10952: Regen docs
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/98ba7037
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/98ba7037
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/98ba7037
Branch: refs/heads/master
Commit: 98ba70378311e3a9612da4c6092ad9efabd69d97
Parents: f3f02f0
Author: Andrea Cosentino <an...@gmail.com>
Authored: Wed Mar 8 08:42:43 2017 +0100
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Wed Mar 8 08:42:43 2017 +0100
----------------------------------------------------------------------
.../src/main/docs/aws-kinesis-firehose-component.adoc | 12 +++++++-----
1 file changed, 7 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/98ba7037/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc b/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
index b691073..adb5c81 100644
--- a/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
+++ b/components/camel-aws/src/main/docs/aws-kinesis-firehose-component.adoc
@@ -35,24 +35,26 @@ The AWS Kinesis Firehose component has no options.
// endpoint options: START
-The AWS Kinesis Firehose component is configured using the URI syntax with the following path and query parameters:
+The AWS Kinesis Firehose endpoint is configured using URI syntax:
aws-kinesis-firehose:streamName
-#### 1 path parameters:
+ with the following path and query parameters:
+
+#### Path Parameters (1 parameters):
[width="100%",cols="2,1,1m,6",options="header"]
|=======================================================================
| Name | Default | Java Type | Description
-| streamName | | String | *Required* Name of the delivery stream
+| streamName | | String | *Required* Name of the stream
|=======================================================================
-#### 5 query parameters:
+#### Query Parameters (2 parameters):
[width="100%",cols="2,1,1m,1m,5",options="header"]
|=======================================================================
| Name | Group | Default | Java Type | Description
-| amazonKinesisFirehoseClient | common | | AmazonKinesisFirehose | *Required* Amazon Kinesis Firehose client to use for all requests for this endpoint
+| amazonKinesisFirehoseClient | producer | | AmazonKinesisFirehose | *Required* Amazon Kinesis Firehose client to use for all requests for this endpoint
| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
|=======================================================================
// endpoint options: END