GH-22: Add KinesisMessageHandler

Fixes GH-22 (https://github.com/spring-projects/spring-integration-aws/issues/22)

* Add `KinesisMessageHandler` based on the `AmazonKinesisAsync.putRecordAsync()` operation.
The logic is fully similar to the `KafkaProducerMessageHandler` since both protocols pursue the same behavior
* Add some Kinesis specific `AwsHeaders`
* Upgrade to Gradle 2.14.1

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) {

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

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<PutRecordRequest, PutRecordResult> asyncHandler;
+
+	private Converter<Object, byte[]> 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<PutRecordRequest, PutRecordResult> asyncHandler) {
+		this.asyncHandler = asyncHandler;
+	}
+
+	public void setConverter(Converter<Object, byte[]> 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<PutRecordResult> 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);
+				}
+			}
+		}
+	}
+
+}

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";
+
 }

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<PutRecordRequest, PutRecordResult> asyncHandler;
+
+	@Test
+	public void testKinesisMessageHandler() {
+		Message<String> 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<PutRecordRequest> 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<PutRecordRequest, PutRecordResult> 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<Object, byte[]>() {
+
+				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;
+		}
+
+	}
+
+}