GH-72: Add KinesisLocalRunning and tests

Fixes spring-projects/spring-integration-aws#72

* Document Kinesis Channel Adapters
* Fix some inconsistency in the `KinesisMessageHandler`
* Add integration test against `KinesisLocalRunning` `@Rule`
* Document testing against Kinesalite

README.md

@@ -504,6 +504,114 @@ The implementation is based on a simple table with `KEY` and `VALUE` attributes,
 By default the `SpringIntegrationMetadataStore` table is used and it is created during `DynamoDbMetaDataStore` initialization if that doesn't exist yet.
 The `DynamoDbMetaDataStore` can be used for the `KinesisMessageDrivenChannelAdapter` as a cloud-based `cehckpointStore`.
 
+For testing application with the `DynamoDbMetaDataStore` you can use [Dynalite][] NPM module.
+What you need in your application is to configure DynamoDB client properly:
+
+````java
+String url = "http://localhost:" + this.port;
+
+this.amazonDynamoDB = AmazonDynamoDBAsyncClientBuilder.standard()
+        .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")))
+        .withClientConfiguration(
+                new ClientConfiguration()
+                        .withMaxErrorRetry(0)
+                        .withConnectionTimeout(1000))
+        .withEndpointConfiguration(
+                new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName()))
+        .build();
+````
+
+Where you should specify the port on which you have ran the Dynalite service.
+Also you can use for you testing purpose a copy of `org.springframework.integration.aws.DynamoDbLocalRunning` in the `/test` directory of this project.    
+
+## Amazon Kinesis
+
+Amazon Kinesis is a platform for streaming data on AWS, making it easy to load and analyze streaming data, and also providing the ability for you to build custom streaming data applications for specialized needs.
+The Spring Integration solution is fully based on the Standard `aws-java-sdk-kinesis` and doesn't use [Kinesis Client Library][] and isn't compatible with it.  
+
+### Inbound Channel Adapter
+
+The `KinesisMessageDrivenChannelAdapter` is an extension of the `MessageProducerSupport` - event-driver channel adapter.
+
+See `KinesisMessageDrivenChannelAdapter` JavaDocs and its setters for more information how to use and how to configure it in the application for Kinesis streams ingestion.
+
+The Java Configuration is pretty simple:
+
+````java
+@SpringBootApplication
+public static class MyConfiguration {
+
+    @Bean
+    public KinesisMessageDrivenChannelAdapter kinesisInboundChannelChannel(AmazonKinesis amazonKinesis) {
+        KinesisMessageDrivenChannelAdapter adapter =
+            new KinesisMessageDrivenChannelAdapter(amazonKinesis, "MY_STREAM");
+        adapter.setOutputChannel(kinesisReceiveChannel());
+        return adapter;
+    }
+}
+````
+
+This channel adapter can be configured with the `DynamoDbMetaDataStore` mentioned above to track sequence checkpoints for shards in the cloud environment when we have several instances of our Kinesis application. 
+By default this adapter uses `DeserializingConverter` to convert `byte[]` from the `Record` data.
+Can be specified as `null` with meaning no conversion and the target `Message` is sent with the `byte[]` payload.
+
+The consumer group is included to the metadata store `key`.
+When records are consumed, they are filtered by the last stored `lastCheckpoint` under the key as `[CONSUMER_GROUP]:[STREAM]:[SHARD_ID]`.
+
+### Outbound Channel Adapter
+
+The `KinesisMessageHandler` is a `AbstractMessageHandler` to perform put record to the Kinesis stream.
+The stream, partition key (or explicit hash key) and sequence number can be determined against request message via evaluation provided expressions or can be specified statically.
+They also can specified as `AwsHeaders.STREAM`, `AwsHeaders.PARTITION_KEY` and `AwsHeaders.SEQUENCE_NUMBER` respectively.
+
+The `payload` of request message can be:
+ 
+- `PutRecordsRequest` to perform `AmazonKinesisAsync.putRecordsAsync`
+- `PutRecordRequest` to perform `AmazonKinesisAsync.putRecordAsync`
+- `ByteBuffer` to represent a data of the `PutRecordRequest`
+- `byte[]` which is wrapped to the `ByteBuffer`
+- any other type which is converted to the `byte[]` by the provided `Converter`; the `SerializingConverter` is used by default.  
+
+The Java Configuration for the message handler:
+
+````java
+@SpringBootApplication
+public static class MyConfiguration {
+
+    @Bean
+    @ServiceActivator(inputChannel = "kinesisSendChannel")
+    public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis) {
+        KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis);
+        kinesisMessageHandler.setPartitionKey("1");
+        return kinesisMessageHandler;
+    }
+    
+}
+````
+
+For testing application with the Kinesis Channel Adapters you can use [Kinesalite][] NPM module.
+What you need in your application is to configure Kinesis client properly:
+
+````java
+String url = "http://localhost:" + this.port;
+
+// See https://github.com/mhart/kinesalite#cbor-protocol-issues-with-the-java-sdk
+System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true");
+
+this.amazonKinesis = AmazonKinesisAsyncClientBuilder.standard()
+        .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")))
+        .withClientConfiguration(
+                new ClientConfiguration()
+                        .withMaxErrorRetry(0)
+                        .withConnectionTimeout(1000))
+        .withEndpointConfiguration(
+                new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName()))
+        .build();
+````
+
+Where you should specify the port on which you have ran the Kinesalite service.
+Also you can use for you testing purpose a copy of `org.springframework.integration.aws.KinesisLocalRunning` in the `/test` directory of this project.    
+
 [Spring Cloud AWS]: https://github.com/spring-cloud/spring-cloud-aws
 [AWS SDK for Java]: http://aws.amazon.com/sdkforjava/
 [Amazon Web Services]: http://aws.amazon.com/
@@ -513,3 +621,6 @@ The `DynamoDbMetaDataStore` can be used for the `KinesisMessageDrivenChannelAdap
 [Reference Manual]: http://docs.spring.io/spring-integration/reference/html/ftp.html
 [Pull requests]: http://help.github.com/send-pull-requests
 [contributor guidelines]: https://github.com/spring-projects/spring-integration/blob/master/CONTRIBUTING.adoc
+[Dynalite]: [https://github.com/mhart/dynalite]
+[Kinesis Client Library]: [https://github.com/awslabs/amazon-kinesis-client]
+[Kinesalite]: [https://github.com/mhart/kinesalite]

src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2017 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.
@@ -131,7 +131,7 @@ public class KinesisMessageHandler extends AbstractMessageHandler {
 		this.explicitHashKeyExpression = explicitHashKeyExpression;
 	}
 
-	public void setSequenceNumberString(String sequenceNumberExpression) {
+	public void setSequenceNumberExpressionString(String sequenceNumberExpression) {
 		setSequenceNumberExpression(EXPRESSION_PARSER.parseExpression(sequenceNumberExpression));
 	}
 
@@ -143,6 +143,14 @@ public class KinesisMessageHandler extends AbstractMessageHandler {
 		this.sync = sync;
 	}
 
+	public void setSendTimeout(long sendTimeout) {
+		setSendTimeoutExpression(new ValueExpression<>(sendTimeout));
+	}
+
+	public void setSendTimeoutExpressionString(String sendTimeoutExpression) {
+		setSendTimeoutExpression(EXPRESSION_PARSER.parseExpression(sendTimeoutExpression));
+	}
+
 	public void setSendTimeoutExpression(Expression sendTimeoutExpression) {
 		this.sendTimeoutExpression = sendTimeoutExpression;
 	}
@@ -162,7 +170,6 @@ public class KinesisMessageHandler extends AbstractMessageHandler {
 					(AsyncHandler<PutRecordsRequest, PutRecordsResult>) this.asyncHandler);
 		}
 		else {
-
 			PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest)
 					? (PutRecordRequest) message.getPayload()
 					: buildPutRecordRequest(message);
@@ -227,8 +234,7 @@ public class KinesisMessageHandler extends AbstractMessageHandler {
 							? (byte[]) payload
 							: this.converter.convert(payload);
 
-			data = ByteBuffer.wrap(
-					bytes);
+			data = ByteBuffer.wrap(bytes);
 		}
 
 		return new PutRecordRequest()

src/test/java/org/springframework/integration/aws/DynamoDbLocalRunning.java

@@ -38,13 +38,16 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
 
 /**
+ * The {@link TestWatcher} implementation for local Amazon DynamoDB service.
+ * See https://github.com/mhart/dynalite.
+ *
  * @author Artem Bilan
  *
  * @since 1.1
  */
-public final class DynamoDbRunning extends TestWatcher {
+public final class DynamoDbLocalRunning extends TestWatcher {
 
-	private static Log logger = LogFactory.getLog(DynamoDbRunning.class);
+	private static Log logger = LogFactory.getLog(DynamoDbLocalRunning.class);
 
 	// Static so that we only test once on failure: speeds up test suite
 	private static Map<Integer, Boolean> dynamoDbOnline = new HashMap<>();
@@ -53,7 +56,7 @@ public final class DynamoDbRunning extends TestWatcher {
 
 	private AmazonDynamoDBAsync amazonDynamoDB;
 
-	private DynamoDbRunning(int port) {
+	private DynamoDbLocalRunning(int port) {
 		this.port = port;
 		dynamoDbOnline.put(port, true);
 	}
@@ -89,8 +92,8 @@ public final class DynamoDbRunning extends TestWatcher {
 	}
 
 
-	public static DynamoDbRunning isRunning(int port) {
-		return new DynamoDbRunning(port);
+	public static DynamoDbLocalRunning isRunning(int port) {
+		return new DynamoDbLocalRunning(port);
 	}
 
 }

src/test/java/org/springframework/integration/aws/KinesisLocalRunning.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2017 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;
+
+import static org.junit.Assume.assumeNoException;
+import static org.junit.Assume.assumeTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.SDKGlobalConfiguration;
+import com.amazonaws.SdkClientException;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.kinesis.AmazonKinesisAsync;
+import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder;
+
+/**
+ * The {@link TestWatcher} implementation for local Amazon Kinesis service.
+ * See https://github.com/mhart/kinesalite.
+ *
+ * @author Artem Bilan
+ *
+ * @since 1.1
+ */
+public final class KinesisLocalRunning extends TestWatcher {
+
+	private static Log logger = LogFactory.getLog(KinesisLocalRunning.class);
+
+	// Static so that we only test once on failure: speeds up test suite
+	private static Map<Integer, Boolean> kinesisOnline = new HashMap<>();
+
+	private final int port;
+
+	private AmazonKinesisAsync amazonKinesis;
+
+	private KinesisLocalRunning(int port) {
+		this.port = port;
+		kinesisOnline.put(port, true);
+	}
+
+	public AmazonKinesisAsync getKinesis() {
+		return this.amazonKinesis;
+	}
+
+	@Override
+	public Statement apply(Statement base, Description description) {
+		assumeTrue(kinesisOnline.get(this.port));
+
+		String url = "http://localhost:" + this.port;
+
+		// See https://github.com/mhart/kinesalite#cbor-protocol-issues-with-the-java-sdk
+		System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true");
+
+		this.amazonKinesis = AmazonKinesisAsyncClientBuilder.standard()
+				.withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")))
+				.withClientConfiguration(
+						new ClientConfiguration()
+								.withMaxErrorRetry(0)
+								.withConnectionTimeout(1000))
+				.withEndpointConfiguration(
+						new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName()))
+				.build();
+
+		try {
+			this.amazonKinesis.listStreams();
+		}
+		catch (SdkClientException e) {
+			logger.warn("Tests not running because no Kinesis on " + url, e);
+			assumeNoException(e);
+		}
+
+
+		return new Statement() {
+			public void evaluate() throws Throwable {
+				try {
+					base.evaluate();
+				} finally {
+					System.clearProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY);
+				}
+
+			}
+		};
+	}
+
+	public static KinesisLocalRunning isRunning(int port) {
+		return new KinesisLocalRunning(port);
+	}
+
+}

src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2017 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.kinesis;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Date;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.aws.KinesisLocalRunning;
+import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter;
+import org.springframework.integration.aws.outbound.KinesisMessageHandler;
+import org.springframework.integration.aws.support.AwsHeaders;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 1.1
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class KinesisIntegrationTests {
+
+	@ClassRule
+	public static final KinesisLocalRunning KINESIS_LOCAL_RUNNING = KinesisLocalRunning.isRunning(4567);
+
+	private static final String TEST_STREAM = "TestStream";
+
+	@Autowired
+	private MessageChannel kinesisSendChannel;
+
+	@Autowired
+	private PollableChannel kinesisReceiveChannel;
+
+	@BeforeClass
+	public static void setup() {
+		KINESIS_LOCAL_RUNNING.getKinesis().createStream(TEST_STREAM, 1);
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		KINESIS_LOCAL_RUNNING.getKinesis().deleteStream(TEST_STREAM);
+	}
+
+	@Test
+	public void testKinesisInboundOutbound() {
+		Date now = new Date();
+		this.kinesisSendChannel.send(
+				MessageBuilder.withPayload(now)
+						.setHeader(AwsHeaders.STREAM, TEST_STREAM)
+						.build());
+		Message<?> receive = this.kinesisReceiveChannel.receive(10_000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo(now);
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class TestConfiguration {
+
+		@Bean
+		@ServiceActivator(inputChannel = "kinesisSendChannel")
+		public MessageHandler kinesisMessageHandler() {
+			KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(KINESIS_LOCAL_RUNNING.getKinesis());
+			kinesisMessageHandler.setPartitionKey("1");
+			return kinesisMessageHandler;
+		}
+
+		@Bean
+		public KinesisMessageDrivenChannelAdapter kinesisInboundChannelChannel() {
+			KinesisMessageDrivenChannelAdapter adapter =
+					new KinesisMessageDrivenChannelAdapter(KINESIS_LOCAL_RUNNING.getKinesis(), TEST_STREAM);
+			adapter.setOutputChannel(kinesisReceiveChannel());
+			return adapter;
+		}
+
+		@Bean
+		public PollableChannel kinesisReceiveChannel() {
+			return new QueueChannel();
+		}
+
+	}
+
+}

src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java

@@ -26,7 +26,7 @@ import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
 
-import org.springframework.integration.aws.DynamoDbRunning;
+import org.springframework.integration.aws.DynamoDbLocalRunning;
 import org.springframework.integration.test.util.TestUtils;
 
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
@@ -47,7 +47,7 @@ import com.amazonaws.waiters.WaiterParameters;
 public class DynamoDbMetadataStoreTests {
 
 	@ClassRule
-	public static final DynamoDbRunning DYNAMO_DB_RUNNING = DynamoDbRunning.isRunning(4567);
+	public static final DynamoDbLocalRunning DYNAMO_DB_RUNNING = DynamoDbLocalRunning.isRunning(4567);
 
 	private static final String TEST_TABLE = "testMetadataStore";