From 42ef916c8058d31e9ba60653d77a02c25d33b55a Mon Sep 17 00:00:00 2001
From: Laurentiu Delcea <laurentiu.delcea@muso.com>
Date: Fri, 20 Sep 2019 23:19:35 +0300
Subject: [PATCH] BAEL-3165 Integrate Kinesis with Spring Binder

---
 .../spring-cloud-stream-kinesis/pom.xml       | 48 +++++++++++++++++++
 .../java/com/baeldung/KinesisApplication.java | 30 ++++++++++++
 .../src/main/resources/application.properties | 11 +++++
 .../KinesisApplicationIntegrationTest.java    | 15 ++++++
 .../src/test/resources/application.properties |  2 +
 5 files changed, 106 insertions(+)
 create mode 100644 spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/pom.xml
 create mode 100644 spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/KinesisApplication.java
 create mode 100644 spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/resources/application.properties
 create mode 100644 spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/java/com/baeldung/KinesisApplicationIntegrationTest.java
 create mode 100644 spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/resources/application.properties

diff --git a/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/pom.xml b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/pom.xml
new file mode 100644
index 0000000000..3817f2747a
--- /dev/null
+++ b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/pom.xml
@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>spring-cloud-stream-kinesis</artifactId>
+    <name>spring-cloud-stream-kinesis</name>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.1.8.RELEASE</version>
+        <relativePath/>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream-binder-kinesis</artifactId>
+            <version>${spring-cloud-stream-kinesis-binder.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-kinesis</artifactId>
+            <version>${aws-sdk.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream-test-support</artifactId>
+            <version>${spring-cloud-stream-test.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <aws-sdk.version>1.11.632</aws-sdk.version>
+        <spring-cloud-stream-kinesis-binder.version>1.2.1.RELEASE</spring-cloud-stream-kinesis-binder.version>
+        <spring-cloud-stream-test.version>2.2.1.RELEASE</spring-cloud-stream-test.version>
+    </properties>
+
+</project>
\ No newline at end of file
diff --git a/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/KinesisApplication.java b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/KinesisApplication.java
new file mode 100644
index 0000000000..c863cd8fe2
--- /dev/null
+++ b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/KinesisApplication.java
@@ -0,0 +1,30 @@
+package com.baeldung;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.messaging.support.MessageBuilder;
+
+@SpringBootApplication
+@EnableBinding(Processor.class)
+public class KinesisApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(KinesisApplication.class, args);
+    }
+
+    @Autowired
+    private Processor processor;
+
+    @StreamListener(Processor.INPUT)
+    public void consume(String val) {
+        System.out.println(val);
+    }
+
+    public void produce(String val) {
+        processor.output().send(MessageBuilder.withPayload(val).build());
+    }
+}
\ No newline at end of file
diff --git a/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/resources/application.properties b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/resources/application.properties
new file mode 100644
index 0000000000..1a966c64fb
--- /dev/null
+++ b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/resources/application.properties
@@ -0,0 +1,11 @@
+cloud.aws.credentials.access-key=aws-key
+cloud.aws.credentials.secret-key=aws-secret
+cloud.aws.region.static=eu-central-1
+cloud.aws.stack.auto=false
+
+spring.cloud.stream.bindings.output.destination=myStream
+spring.cloud.stream.bindings.output.content-type=text/plain
+
+spring.cloud.stream.bindings.input.destination=myStream
+spring.cloud.stream.bindings.input.group=myStream-group
+spring.cloud.stream.bindings.input.content-type=text/plain
\ No newline at end of file
diff --git a/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/java/com/baeldung/KinesisApplicationIntegrationTest.java b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/java/com/baeldung/KinesisApplicationIntegrationTest.java
new file mode 100644
index 0000000000..4e1c281cfb
--- /dev/null
+++ b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/java/com/baeldung/KinesisApplicationIntegrationTest.java
@@ -0,0 +1,15 @@
+package com.baeldung;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = KinesisApplication.class)
+public class KinesisApplicationIntegrationTest {
+    @Test
+    public void whenSpringContextIsBootstrapped_thenNoExceptions() {
+
+    }
+}
\ No newline at end of file
diff --git a/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/resources/application.properties b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/resources/application.properties
new file mode 100644
index 0000000000..48005616ec
--- /dev/null
+++ b/spring-cloud/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/resources/application.properties
@@ -0,0 +1,2 @@
+cloud.aws.region.static=eu-central-1
+cloud.aws.stack.auto=false
\ No newline at end of file