This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 40182ce0ea0bf0fef0a0fb12ab3e217f69ed6d71
Author: Andrea Cosentino <anco...@gmail.com>
AuthorDate: Tue Jun 30 14:12:11 2020 +0200
CAMEL-15191 - Fixed CS
---
...KinesisFirehose2ComponentVerifierExtension.java | 3 ++-
.../firehose/KinesisFirehose2Configuration.java | 14 +++++++-------
.../aws2/firehose/KinesisFirehose2Endpoint.java | 3 ++-
.../aws2/firehose/KinesisFirehose2Producer.java | 22 +++++++++++-----------
.../KinesisFirehoseComponentIntegrationTest.java | 17 ++++++++---------
5 files changed, 30 insertions(+), 29 deletions(-)
diff --git
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java
index 45d6f16..14e25f3 100644
---
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java
+++
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java
@@ -68,7 +68,8 @@ public class KinesisFirehose2ComponentVerifierExtension
extends DefaultComponent
try {
KinesisFirehose2Configuration configuration = setProperties(new
KinesisFirehose2Configuration(), parameters);
if
(!FirehoseClient.serviceMetadata().regions().contains(Region.of(configuration.getRegion())))
{
- ResultErrorBuilder errorBuilder =
ResultErrorBuilder.withCodeAndDescription(VerificationError.StandardCode.ILLEGAL_PARAMETER,
"The service is not supported in this region");
+ ResultErrorBuilder errorBuilder =
ResultErrorBuilder.withCodeAndDescription(VerificationError.StandardCode.ILLEGAL_PARAMETER,
+
"The service is not supported in this region");
return builder.error(errorBuilder.build()).build();
}
AwsBasicCredentials cred =
AwsBasicCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey());
diff --git
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
index 03a5693..3c6b5d2 100644
---
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
+++
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
@@ -113,18 +113,18 @@ public class KinesisFirehose2Configuration implements
Cloneable {
}
public KinesisFirehose2Operations getOperation() {
- return operation;
- }
+ return operation;
+ }
+
+ public void setOperation(KinesisFirehose2Operations operation) {
+ this.operation = operation;
+ }
- public void setOperation(KinesisFirehose2Operations operation) {
- this.operation = operation;
- }
-
// *************************************************
//
// *************************************************
- public KinesisFirehose2Configuration copy() {
+ public KinesisFirehose2Configuration copy() {
try {
return (KinesisFirehose2Configuration)super.clone();
} catch (CloneNotSupportedException e) {
diff --git
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
index 6e427ce..d4c76ea 100644
---
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
+++
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
@@ -37,7 +37,8 @@ import
software.amazon.awssdk.services.firehose.FirehoseClientBuilder;
/**
* Produce data to AWS Kinesis Firehose streams using AWS SDK version 2.x.
*/
-@UriEndpoint(firstVersion = "3.2.0", scheme = "aws2-kinesis-firehose", title =
"AWS 2 Kinesis Firehose", syntax = "aws2-kinesis-firehose:streamName",
producerOnly = true, category = {Category.CLOUD, Category.MESSAGING})
+@UriEndpoint(firstVersion = "3.2.0", scheme = "aws2-kinesis-firehose", title =
"AWS 2 Kinesis Firehose", syntax = "aws2-kinesis-firehose:streamName",
producerOnly = true, category = {Category.CLOUD,
+
Category.MESSAGING})
public class KinesisFirehose2Endpoint extends DefaultEndpoint {
@UriParam
diff --git
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
index 26b8abc..6792bdb 100644
---
a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
+++
b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
@@ -61,25 +61,25 @@ public class KinesisFirehose2Producer extends
DefaultProducer {
}
}
}
-
+
private void sendBatchRecord(FirehoseClient client, Exchange exchange) {
if (exchange.getIn().getBody() instanceof Iterable) {
Iterable c = exchange.getIn().getBody(Iterable.class);
PutRecordBatchRequest.Builder batchRequest =
PutRecordBatchRequest.builder();
batchRequest.deliveryStreamName(getEndpoint().getConfiguration().getStreamName());
- batchRequest.records((Collection<Record>) c);
+ batchRequest.records((Collection<Record>)c);
PutRecordBatchResponse result =
client.putRecordBatch(batchRequest.build());
Message message = getMessageForResponse(exchange);
message.setBody(result);
} else {
- PutRecordBatchRequest req =
exchange.getIn().getBody(PutRecordBatchRequest.class);
- PutRecordBatchResponse result = client.putRecordBatch(req);
+ PutRecordBatchRequest req =
exchange.getIn().getBody(PutRecordBatchRequest.class);
+ PutRecordBatchResponse result = client.putRecordBatch(req);
Message message = getMessageForResponse(exchange);
message.setBody(result);
- }
- }
+ }
+ }
- public void processSingleRecord(final Exchange exchange) {
+ public void processSingleRecord(final Exchange exchange) {
PutRecordRequest request = createRequest(exchange);
LOG.trace("Sending request [{}] from exchange [{}]...", request,
exchange);
PutRecordResponse putRecordResult =
getEndpoint().getClient().putRecord(request);
@@ -102,17 +102,17 @@ public class KinesisFirehose2Producer extends
DefaultProducer {
public static Message getMessageForResponse(final Exchange exchange) {
return exchange.getMessage();
}
-
+
protected FirehoseClient getClient() {
return getEndpoint().getClient();
}
-
+
protected KinesisFirehose2Configuration getConfiguration() {
return getEndpoint().getConfiguration();
}
-
+
private KinesisFirehose2Operations determineOperation(Exchange exchange) {
- KinesisFirehose2Operations operation =
exchange.getIn().getHeader(KinesisFirehose2Constants.KINESIS_FIREHOSE_OPERATION,
KinesisFirehose2Operations.class);
+ KinesisFirehose2Operations operation =
exchange.getIn().getHeader(KinesisFirehose2Constants.KINESIS_FIREHOSE_OPERATION,
KinesisFirehose2Operations.class);
if (operation == null) {
operation = getConfiguration().getOperation();
}
diff --git
a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
index 8b39e6f..b971a37 100644
---
a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
+++
b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.aws2.firehose.integration;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.camel.BindToRegistry;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -25,17 +29,12 @@ import
org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants;
import org.apache.camel.test.junit5.CamelTestSupport;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
-
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.firehose.FirehoseClient;
import software.amazon.awssdk.services.firehose.model.Record;
import static org.junit.jupiter.api.Assertions.assertNotNull;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.List;
-
@Disabled("Must be manually tested.")
public class KinesisFirehoseComponentIntegrationTest extends CamelTestSupport {
@@ -56,12 +55,12 @@ public class KinesisFirehoseComponentIntegrationTest
extends CamelTestSupport {
public void testFirehoseBatchRouting() throws Exception {
Exchange exchange = template.send("direct:start",
ExchangePattern.InOnly, new Processor() {
public void process(Exchange exchange) throws Exception {
- List<Record> recs = new ArrayList<Record>();
- Record rec =
Record.builder().data(SdkBytes.fromString(("Test1"),
Charset.defaultCharset())).build();
- Record rec1 =
Record.builder().data(SdkBytes.fromString(("Test2"),
Charset.defaultCharset())).build();
+ List<Record> recs = new ArrayList<Record>();
+ Record rec =
Record.builder().data(SdkBytes.fromString("Test1",
Charset.defaultCharset())).build();
+ Record rec1 =
Record.builder().data(SdkBytes.fromString("Test2",
Charset.defaultCharset())).build();
recs.add(rec);
recs.add(rec1);
- exchange.getIn().setBody(recs);
+ exchange.getIn().setBody(recs);
}
});
assertNotNull(exchange.getIn().getBody());
