diff --git a/build.gradle b/build.gradle index 7e4254e..9372e43 100644 --- a/build.gradle +++ b/build.gradle @@ -12,7 +12,7 @@ plugins { id 'eclipse' id 'idea' id 'jacoco' - id 'org.sonarqube' version '1.2' + id 'org.sonarqube' version '2.1' id 'checkstyle' } description = 'Spring Integration AWS Support' @@ -52,10 +52,11 @@ if (project.hasProperty('platformVersion')) { ext { assertjVersion = '3.5.2' + awsKinesisVersion = '1.7.0' servletApiVersion = '3.1.0' slf4jVersion = '1.7.21' springCloudAwsVersion = '1.1.1.RELEASE' - springIntegrationVersion = '4.3.2.RELEASE' + springIntegrationVersion = '4.3.4.RELEASE' idPrefix = 'aws' @@ -81,12 +82,15 @@ checkstyle { } dependencies { - compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion" compile "org.springframework.cloud:spring-cloud-aws-core:$springCloudAwsVersion" + compile("org.springframework.cloud:spring-cloud-aws-messaging:$springCloudAwsVersion", optional) compile("org.springframework.integration:spring-integration-file:$springIntegrationVersion", optional) compile("org.springframework.integration:spring-integration-http:$springIntegrationVersion", optional) + + compile("com.amazonaws:amazon-kinesis-client:$awsKinesisVersion", optional) + compile("javax.servlet:javax.servlet-api:$servletApiVersion", provided) testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion" @@ -153,6 +157,7 @@ jacocoTestReport { } } +check.dependsOn javadoc build.dependsOn jacocoTestReport task sourcesJar(type: Jar) { diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar index d3b8398..3baa851 100644 Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index eaf9f7d..9bfaaef 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,6 +1,6 @@ -#Tue Sep 20 11:41:48 EDT 2016 +#Wed Oct 12 19:15:34 EDT 2016 distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-2.14-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-2.14.1-bin.zip diff --git a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java new file mode 100644 index 0000000..149799d --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java @@ -0,0 +1,200 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed 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.springframework.integration.aws.outbound; + +import java.nio.ByteBuffer; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.springframework.core.convert.converter.Converter; +import org.springframework.core.serializer.support.SerializingConverter; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +import com.amazonaws.handlers.AsyncHandler; +import com.amazonaws.services.kinesis.AmazonKinesisAsync; +import com.amazonaws.services.kinesis.model.PutRecordRequest; +import com.amazonaws.services.kinesis.model.PutRecordResult; + +/** + * The {@link AbstractMessageHandler} implementation for the Amazon Kinesis {@code putRecord(s)}. + * + * @author Artem Bilan + * @since 1.1 + * + * @see AmazonKinesisAsync#putRecord(PutRecordRequest) + * @see com.amazonaws.handlers.AsyncHandler + */ +public class KinesisMessageHandler extends AbstractMessageHandler { + + private static final long DEFAULT_SEND_TIMEOUT = 10000; + + private final AmazonKinesisAsync amazonKinesis; + + private AsyncHandler asyncHandler; + + private Converter converter = new SerializingConverter(); + + private EvaluationContext evaluationContext; + + private volatile Expression streamExpression; + + private volatile Expression partitionKeyExpression; + + private volatile Expression explicitHashKeyExpression; + + private volatile Expression sequenceNumberExpression; + + private boolean sync; + + private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT); + + public KinesisMessageHandler(AmazonKinesisAsync amazonKinesis) { + this.amazonKinesis = amazonKinesis; + } + + public void setAsyncHandler(AsyncHandler asyncHandler) { + this.asyncHandler = asyncHandler; + } + + public void setConverter(Converter converter) { + Assert.notNull(converter, "'converter' must not be null."); + this.converter = converter; + } + + public void setStream(String stream) { + setStreamExpression(new LiteralExpression(stream)); + } + + public void setStreamExpressionString(String streamExpression) { + setStreamExpression(EXPRESSION_PARSER.parseExpression(streamExpression)); + } + + public void setStreamExpression(Expression streamExpression) { + this.streamExpression = streamExpression; + } + + public void setPartitionKey(String partitionKey) { + setPartitionKeyExpression(new LiteralExpression(partitionKey)); + } + + public void setPartitionKeyExpressionString(String partitionKeyExpression) { + setPartitionKeyExpression(EXPRESSION_PARSER.parseExpression(partitionKeyExpression)); + } + + public void setPartitionKeyExpression(Expression partitionKeyExpression) { + this.partitionKeyExpression = partitionKeyExpression; + } + + public void setExplicitHashKey(String explicitHashKey) { + setExplicitHashKeyExpression(new LiteralExpression(explicitHashKey)); + } + + public void setExplicitHashKeyExpressionString(String explicitHashKeyExpression) { + setExplicitHashKeyExpression(EXPRESSION_PARSER.parseExpression(explicitHashKeyExpression)); + } + + public void setExplicitHashKeyExpression(Expression explicitHashKeyExpression) { + this.explicitHashKeyExpression = explicitHashKeyExpression; + } + + public void setSequenceNumberString(String sequenceNumberExpression) { + setSequenceNumberExpression(EXPRESSION_PARSER.parseExpression(sequenceNumberExpression)); + } + + public void setSequenceNumberExpression(Expression sequenceNumberExpression) { + this.sequenceNumberExpression = sequenceNumberExpression; + } + + public void setSync(boolean sync) { + this.sync = sync; + } + + public void setSendTimeoutExpression(Expression sendTimeoutExpression) { + this.sendTimeoutExpression = sendTimeoutExpression; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + } + + @Override + protected void handleMessageInternal(Message message) throws Exception { + String stream = message.getHeaders().get(AwsHeaders.STREAM, String.class); + if (!StringUtils.hasText(stream) && this.streamExpression != null) { + stream = this.streamExpression.getValue(this.evaluationContext, message, String.class); + } + Assert.state(stream != null, "'stream' must not be null for sending a Kinesis record. " + + "Consider configuring this handler with a 'stream'( or 'streamExpression') or supply an " + + "'aws_stream' message header."); + + String partitionKey = message.getHeaders().get(AwsHeaders.PARTITION_KEY, String.class); + if (!StringUtils.hasText(partitionKey) && this.partitionKeyExpression != null) { + partitionKey = this.partitionKeyExpression.getValue(this.evaluationContext, message, String.class); + } + Assert.state(partitionKey != null, "'partitionKey' must not be null for sending a Kinesis record. " + + "Consider configuring this handler with a 'partitionKey'( or 'partitionKeyExpression') or supply an " + + "'aws_partitionKey' message header."); + + String explicitHashKey = + (this.explicitHashKeyExpression != null + ? this.explicitHashKeyExpression.getValue(this.evaluationContext, message, String.class) + : null); + + String sequenceNumber = message.getHeaders().get(AwsHeaders.SEQUENCE_NUMBER, String.class); + if (!StringUtils.hasText(stream) && this.streamExpression != null) { + partitionKey = this.sequenceNumberExpression.getValue(this.evaluationContext, message, String.class); + } + + PutRecordRequest putRecordRequest = new PutRecordRequest() + .withStreamName(stream) + .withPartitionKey(partitionKey) + .withExplicitHashKey(explicitHashKey) + .withSequenceNumberForOrdering(sequenceNumber) + .withData(ByteBuffer.wrap(this.converter.convert(message.getPayload()))); + + Future resultFuture = this.amazonKinesis.putRecordAsync(putRecordRequest, this.asyncHandler); + + if (this.sync) { + Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class); + if (sendTimeout == null || sendTimeout < 0) { + resultFuture.get(); + } + else { + try { + resultFuture.get(sendTimeout, TimeUnit.MILLISECONDS); + } + catch (TimeoutException te) { + throw new MessageTimeoutException(message, "Timeout waiting for response from AmazonKinesis", te); + } + } + } + } + +} diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java index d9ca874..61a90d8 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -65,4 +65,19 @@ public abstract class AwsHeaders { */ public static final String SNS_PUBLISHED_MESSAGE_ID = PREFIX + "snsPublishedMessageId"; + /** + * The {@value STREAM} header for sending/receiving data over Kinesis. + */ + public static final String STREAM = PREFIX + "stream"; + + /** + * The {@value PARTITION_KEY} header for sending/receiving data over Kinesis. + */ + public static final String PARTITION_KEY = PREFIX + "partitionKey"; + + /** + * The {@value SEQUENCE_NUMBER} header for sending/receiving data over Kinesis. + */ + public static final String SEQUENCE_NUMBER = PREFIX + "sequenceNumber"; + } diff --git a/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java new file mode 100644 index 0000000..f6905e4 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java @@ -0,0 +1,160 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed 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.springframework.integration.aws.outbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.nio.ByteBuffer; +import java.util.concurrent.Future; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.convert.converter.Converter; +import org.springframework.core.serializer.support.SerializingConverter; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.context.junit4.SpringRunner; + +import com.amazonaws.handlers.AsyncHandler; +import com.amazonaws.services.kinesis.AmazonKinesisAsync; +import com.amazonaws.services.kinesis.model.PutRecordRequest; +import com.amazonaws.services.kinesis.model.PutRecordResult; + +/** + * @author Artem Bilan + * @since 1.1 + */ +@RunWith(SpringRunner.class) +public class KinesisMessageHandlerTests { + + @Autowired + protected AmazonKinesisAsync amazonKinesis; + + @Autowired + protected MessageChannel kinesisSendChannel; + + @Autowired + protected KinesisMessageHandler kinesisMessageHandler; + + @Autowired + protected AsyncHandler asyncHandler; + + @Test + public void testKinesisMessageHandler() { + Message message = MessageBuilder.withPayload("message").build(); + try { + this.kinesisSendChannel.send(message); + } + catch (Exception e) { + assertThat(e).isInstanceOf(MessageHandlingException.class); + assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(e.getMessage()).contains("'stream' must not be null for sending a Kinesis record"); + } + + this.kinesisMessageHandler.setStream("foo"); + try { + this.kinesisSendChannel.send(message); + } + catch (Exception e) { + assertThat(e).isInstanceOf(MessageHandlingException.class); + assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(e.getMessage()).contains("'partitionKey' must not be null for sending a Kinesis record"); + } + + message = MessageBuilder.fromMessage(message) + .setHeader(AwsHeaders.PARTITION_KEY, "fooKey") + .setHeader(AwsHeaders.SEQUENCE_NUMBER, "10") + .build(); + + this.kinesisSendChannel.send(message); + + ArgumentCaptor putRecordRequestArgumentCaptor = + ArgumentCaptor.forClass(PutRecordRequest.class); + verify(this.amazonKinesis).putRecordAsync(putRecordRequestArgumentCaptor.capture(), eq(this.asyncHandler)); + + PutRecordRequest putRecordRequest = putRecordRequestArgumentCaptor.getValue(); + + assertThat(putRecordRequest.getStreamName()).isEqualTo("foo"); + assertThat(putRecordRequest.getPartitionKey()).isEqualTo("fooKey"); + assertThat(putRecordRequest.getSequenceNumberForOrdering()).isEqualTo("10"); + assertThat(putRecordRequest.getExplicitHashKey()).isNull(); + assertThat(putRecordRequest.getData()).isEqualTo(ByteBuffer.wrap("message".getBytes())); + } + + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + @SuppressWarnings("unchecked") + public AmazonKinesisAsync amazonKinesis() { + AmazonKinesisAsync mock = mock(AmazonKinesisAsync.class); + given(mock.putRecordAsync(any(PutRecordRequest.class), any(AsyncHandler.class))) + .willReturn(mock(Future.class)); + return mock; + } + + @Bean + @SuppressWarnings("unchecked") + public AsyncHandler asyncHandler() { + return mock(AsyncHandler.class); + } + + @Bean + @ServiceActivator(inputChannel = "kinesisSendChannel") + public MessageHandler kinesisMessageHandler() { + KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis()); + kinesisMessageHandler.setSync(true); + kinesisMessageHandler.setAsyncHandler(asyncHandler()); + kinesisMessageHandler.setConverter(new Converter() { + + private SerializingConverter serializingConverter = new SerializingConverter(); + + @Override + public byte[] convert(Object source) { + if (source instanceof String) { + return ((String) source).getBytes(); + } + else { + return this.serializingConverter.convert(source); + } + } + + }); + return kinesisMessageHandler; + } + + } + +}