JAVA-14987 Added missing code and split the code in packages (#13238)

- Split the code in respective packages
- Added missing codes

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/pom.xml

@@ -17,11 +17,6 @@
                <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>
@@ -33,6 +28,21 @@
                <version>${spring-cloud-stream-test.version}</version>
                <scope>test</scope>
           </dependency>
+          <dependency>
+               <groupId>com.amazonaws</groupId>
+               <artifactId>amazon-kinesis-producer</artifactId>
+               <version>0.13.1</version>
+          </dependency>
+          <dependency>
+               <groupId>com.amazonaws</groupId>
+               <artifactId>amazon-kinesis-client</artifactId>
+               <version>1.11.2</version>
+          </dependency>
+          <dependency>
+               <groupId>org.springframework.cloud</groupId>
+               <artifactId>spring-cloud-stream-binder-kinesis</artifactId>
+               <version>${spring-cloud-stream-kinesis-binder.version}</version>
+          </dependency>
      </dependencies>
 
      <properties>

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/KinesisApplication.java

@@ -1,53 +0,0 @@
-package com.baeldung;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-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.context.annotation.Bean;
-import org.springframework.messaging.support.MessageBuilder;
-
-import com.amazonaws.auth.AWSStaticCredentialsProvider;
-import com.amazonaws.auth.BasicAWSCredentials;
-import com.amazonaws.regions.Regions;
-import com.amazonaws.services.kinesis.AmazonKinesis;
-import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
-
-@SpringBootApplication
-@EnableBinding(Processor.class)
-public class KinesisApplication {
-
-    @Value("${aws.access.key}")
-    private String accessKey;
-    
-    @Value("${aws.secret.key}")
-    private String secretKey;
-    
-    @Autowired
-    private Processor processor;
-
-    public static void main(String[] args) {
-        SpringApplication.run(KinesisApplication.class, args);
-    }
-
-    @Bean
-    public AmazonKinesis buildAmazonKinesis() {
-        BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
-        return AmazonKinesisClientBuilder.standard()
-            .withCredentials(new AWSStaticCredentialsProvider(awsCredentials))
-            .withRegion(Regions.EU_CENTRAL_1)
-            .build();
-    }
-
-    @StreamListener(Processor.INPUT)
-    public void consume(String val) {
-        System.out.println(val);
-    }
-
-    public void produce(String val) {
-        processor.output().send(MessageBuilder.withPayload(val).build());
-    }
-}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/binder/ConsumerBinder.java

@@ -0,0 +1,16 @@
+package com.baeldung.binder;
+
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.stereotype.Component;
+
+@Component
+@EnableBinding(Sink.class)
+public class ConsumerBinder {
+
+	@StreamListener(Sink.INPUT)
+	public void consume(String ip) {
+		System.out.println(ip);
+	}
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/binder/KinesisBinderApplication.java

@@ -0,0 +1,12 @@
+package com.baeldung.binder;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class KinesisBinderApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(KinesisBinderApplication.class, args);
+	}
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/binder/ProducerBinder.java

@@ -0,0 +1,24 @@
+package com.baeldung.binder;
+
+import java.util.stream.IntStream;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+@Component
+@EnableBinding(Source.class)
+public class ProducerBinder {
+
+    @Autowired
+    private Source source;
+
+    @Scheduled(fixedDelay = 3000L)
+    private void produce() {
+        IntStream.range(1, 200).mapToObj(ipSuffix -> "192.168.0." + ipSuffix)
+            .forEach(entry -> source.output().send(MessageBuilder.withPayload(entry).build()));
+    }
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/IpProcessor.java

@@ -1,4 +1,4 @@
-package com.baeldung;
+package com.baeldung.kclkpl;
 
 import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
 import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/IpProcessorFactory.java

@@ -1,4 +1,4 @@
-package com.baeldung;
+package com.baeldung.kclkpl;
 
 import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
 import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/IpProducer.java

@@ -0,0 +1,30 @@
+package com.baeldung.kclkpl;
+
+import java.nio.ByteBuffer;
+import java.util.stream.IntStream;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import com.amazonaws.services.kinesis.producer.KinesisProducer;
+
+@Component
+public class IpProducer {
+
+    @Value("${ips.stream}")
+    private String IPS_STREAM;
+    
+    @Value("${ips.partition.key}")
+    private String IPS_PARTITION_KEY;
+    
+    @Autowired
+    private KinesisProducer kinesisProducer;
+
+    @Scheduled(fixedDelay = 3000L)
+    private void produce() {
+        IntStream.range(1, 200).mapToObj(ipSuffix -> ByteBuffer.wrap(("192.168.0." + ipSuffix).getBytes()))
+            .forEach(entry -> kinesisProducer.addUserRecord(IPS_STREAM, IPS_PARTITION_KEY, entry));
+    }
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/KinesisKCLApplication.java

@@ -0,0 +1,48 @@
+package com.baeldung.kclkpl;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;
+
+@SpringBootApplication
+public class KinesisKCLApplication implements ApplicationRunner {
+
+    @Value("${aws.access.key}")
+    private String accessKey;
+    
+    @Value("${aws.secret.key}")
+    private String secretKey;
+    
+    @Value("${ips.stream}")
+    private String IPS_STREAM;
+    
+    public static void main(String[] args) {
+        SpringApplication.run(KinesisKCLApplication.class, args);
+    }
+
+	@Override
+	public void run(ApplicationArguments args) throws Exception {
+		BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
+		KinesisClientLibConfiguration consumerConfig = new KinesisClientLibConfiguration(
+		  "KinesisKCLConsumer", 
+		  IPS_STREAM,
+		  new AWSStaticCredentialsProvider(awsCredentials), 
+		  "KinesisKCLConsumer")
+		    .withRegionName(Regions.EU_CENTRAL_1.getName());
+
+		new Worker.Builder()
+		  .recordProcessorFactory(new IpProcessorFactory())
+		  .config(consumerConfig)
+		  .build()
+		  .run();
+	}
+    
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/kclkpl/KinesisKPLApplication.java

@@ -0,0 +1,38 @@
+package com.baeldung.kclkpl;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.kinesis.producer.KinesisProducer;
+import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration;
+
+@SpringBootApplication
+public class KinesisKPLApplication {
+
+    @Value("${aws.access.key}")
+    private String accessKey;
+    
+    @Value("${aws.secret.key}")
+    private String secretKey;
+    
+    public static void main(String[] args) {
+        SpringApplication.run(KinesisKPLApplication.class, args);
+    }
+
+    @Bean
+    public KinesisProducer kinesisProducer() {
+    	BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
+    	KinesisProducerConfiguration producerConfig = new KinesisProducerConfiguration()
+    	  .setCredentialsProvider(new AWSStaticCredentialsProvider(awsCredentials))
+    	  .setVerifyCertificate(false)
+    	  .setRegion(Regions.EU_CENTRAL_1.getName());
+
+    	return new KinesisProducer(producerConfig);
+    }
+    
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/sdk/ConsumerSDK.java

@@ -1,12 +1,9 @@
-package com.baeldung;
+package com.baeldung.sdk;
 
 import javax.annotation.PostConstruct;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.annotation.StreamListener;
-import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.stereotype.Component;
 
 import com.amazonaws.services.kinesis.AmazonKinesis;
@@ -17,8 +14,7 @@ import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
 import com.amazonaws.services.kinesis.model.ShardIteratorType;
 
 @Component
-@EnableBinding(Sink.class)
+public class ConsumerSDK {
-public class IpConsumer {
 
     @Value("${ips.stream}")
     private String IPS_STREAM;
@@ -31,12 +27,7 @@ public class IpConsumer {
     
     private GetShardIteratorResult shardIterator;
 
-    @StreamListener(Sink.INPUT)
+    public void consumeWithKinesis() {
-    public void consume(String ip) {
-        System.out.println(ip);
-    }
-
-    private void consumeWithKinesis() {
         GetRecordsRequest recordsRequest = new GetRecordsRequest();
         recordsRequest.setShardIterator(shardIterator.getShardIterator());
         recordsRequest.setLimit(25);

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/sdk/KinesisSDKApplication.java

@@ -0,0 +1,35 @@
+package com.baeldung.sdk;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
+
+@SpringBootApplication
+public class KinesisSDKApplication {
+
+	@Value("${aws.access.key}")
+	private String accessKey;
+
+	@Value("${aws.secret.key}")
+	private String secretKey;
+
+	public static void main(String[] args) {
+		SpringApplication.run(KinesisSDKApplication.class, args);
+	}
+
+	@Bean
+	public AmazonKinesis buildAmazonKinesis() {
+		BasicAWSCredentials awsCredentials = new BasicAWSCredentials(accessKey, secretKey);
+		return AmazonKinesisClientBuilder.standard()
+		    .withCredentials(new AWSStaticCredentialsProvider(awsCredentials))
+		    .withRegion(Regions.EU_CENTRAL_1)
+		    .build();
+	}
+}

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/java/com/baeldung/sdk/ProducerSDK.java

@@ -1,4 +1,4 @@
-package com.baeldung;
+package com.baeldung.sdk;
 
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -7,9 +7,6 @@ import java.util.stream.IntStream;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.messaging.Source;
-import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -18,8 +15,7 @@ import com.amazonaws.services.kinesis.model.PutRecordsRequest;
 import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry;
 
 @Component
-@EnableBinding(Source.class)
+public class ProducerSDK {
-public class IpProducer {
 
     @Value("${ips.partition.key}")
     private String IPS_PARTITION_KEY;
@@ -27,17 +23,9 @@ public class IpProducer {
     @Value("${ips.stream}")
     private String IPS_STREAM;
 
-    @Autowired
-    private Source source;
     @Autowired
     private AmazonKinesis kinesis;
 
-    @Scheduled(fixedDelay = 3000L)
-    private void produce() {
-        IntStream.range(1, 200).mapToObj(ipSuffix -> "192.168.0." + ipSuffix)
-            .forEach(entry -> source.output().send(MessageBuilder.withPayload(entry).build()));
-    }
-
     @Scheduled(fixedDelay = 3000L)
     private void produceWithKinesis() {
         List<PutRecordsRequestEntry> entries = IntStream.range(1, 200).mapToObj(ipSuffix -> {

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/main/resources/application.properties

@@ -1,6 +1,12 @@
+# configurations for AWS SDK consumer and producer
 aws.access.key=my-aws-access-key-goes-here
 aws.secret.key=my-aws-secret-key-goes-here
 
+ips.partition.key=ips-partition-key
+ips.stream=ips-stream
+ips.shard.id=1
+
+# configurations for Spring Cloud Stream Kineses Binder consumer and producer
 cloud.aws.credentials.access-key=my-aws-access-key
 cloud.aws.credentials.secret-key=my-aws-secret-key
 cloud.aws.region.static=eu-central-1
@@ -12,7 +18,3 @@ spring.cloud.stream.bindings.input.content-type=text/plain
 
 spring.cloud.stream.bindings.output.destination=myStream
 spring.cloud.stream.bindings.output.content-type=text/plain
-
-ips.partition.key=ips-partition-key
-ips.stream=ips-stream
-ips.shard.id=1

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/java/com/baeldung/KinesisApplicationManualTest.java

@@ -5,11 +5,13 @@ import org.junit.runner.RunWith;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import com.baeldung.kclkpl.KinesisKPLApplication;
+
 /**
  * Manual Test - this test needs correct AWS Access Key and Secret to build the Amazon Kinesis and complete successfully
  */
 @RunWith(SpringRunner.class)
-@SpringBootTest(classes = KinesisApplication.class)
+@SpringBootTest(classes = KinesisKPLApplication.class)
 public class KinesisApplicationManualTest {
 	@Test
 	public void whenSpringContextIsBootstrapped_thenNoExceptions() {

spring-cloud-modules/spring-cloud-stream/spring-cloud-stream-kinesis/src/test/resources/application.properties

@@ -1,6 +1,12 @@
+# configurations for AWS SDK consumer and producer
 aws.access.key=my-aws-access-key-goes-here
 aws.secret.key=my-aws-secret-key-goes-here
 
+ips.partition.key=ips-partition-key
+ips.stream=ips-stream
+ips.shard.id=1
+
+# configurations for Spring Cloud Stream Kineses Binder consumer and producer
 cloud.aws.credentials.access-key=my-aws-access-key
 cloud.aws.credentials.secret-key=my-aws-secret-key
 cloud.aws.region.static=eu-central-1
@@ -12,7 +18,3 @@ spring.cloud.stream.bindings.input.content-type=text/plain
 
 spring.cloud.stream.bindings.output.destination=myStream
 spring.cloud.stream.bindings.output.content-type=text/plain
-
-ips.partition.key=ips-partition-key
-ips.stream=ips-stream
-ips.shard.id=1