adds integration test

aws-kinesis-project/producer/.localstack/init-aws.sh

@@ -1,4 +1,4 @@
-#!/usr/bin/env bash
+#!/bin/bash
 
 awslocal kinesis create-stream --stream-name my-test-stream --shard-count 1
 

aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java

@@ -1,10 +1,38 @@
 package com.learning.aws.spring;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.learning.aws.spring.common.AbstractIntegrationTest;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
 import org.junit.jupiter.api.Test;
+import org.springframework.integration.aws.support.AwsHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
 
 class ApplicationIntegrationTest extends AbstractIntegrationTest {
 
     @Test
-    void contextLoads() {}
+    void contextLoads() throws InterruptedException, JsonProcessingException {
+
+        assertThat(this.messageBarrier.await(30, TimeUnit.SECONDS)).isTrue();
+
+        Message<List<?>> message = this.messageHolder.get();
+        assertThat(message.getHeaders())
+                .containsKeys(AwsHeaders.CHECKPOINTER, AwsHeaders.SHARD, AwsHeaders.RECEIVED_STREAM)
+                .doesNotContainKeys(AwsHeaders.STREAM, AwsHeaders.PARTITION_KEY);
+
+        List<?> payload = message.getPayload();
+        assertThat(payload).hasSize(10);
+
+        Object item = payload.get(0);
+
+        assertThat(item).isInstanceOf(GenericMessage.class);
+
+        Message<?> messageFromBatch = (Message<?>) item;
+
+        assertThat(messageFromBatch.getPayload()).isEqualTo("Message0");
+        assertThat(messageFromBatch.getHeaders()).containsEntry("event.eventType", "createEvent");
+    }
 }

aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/TestKinesisProducerApplication.java

@@ -2,9 +2,14 @@
 
 import static org.testcontainers.containers.localstack.LocalStackContainer.Service.KINESIS;
 
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
 import org.springframework.test.context.DynamicPropertyRegistry;
 import org.testcontainers.containers.localstack.LocalStackContainer;
 import org.testcontainers.utility.DockerImageName;
@@ -44,6 +49,24 @@ KinesisAsyncClient amazonKinesis(LocalStackContainer localStackContainer) {
                 .build();
     }
 
+    @Bean
+    public AtomicReference<Message<List<Message<?>>>> messageHolder() {
+        return new AtomicReference<>();
+    }
+
+    @Bean
+    public CountDownLatch messageBarrier() {
+        return new CountDownLatch(1);
+    }
+
+    @Bean
+    public Consumer<Message<List<Message<?>>>> eventConsumerBatchProcessingWithHeaders() {
+        return eventMessages -> {
+            messageHolder().set(eventMessages);
+            messageBarrier().countDown();
+        };
+    }
+
     public static void main(String[] args) {
         SpringApplication.from(KinesisProducerApplication::main)
                 .with(TestKinesisProducerApplication.class)

aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java

@@ -5,18 +5,39 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.learning.aws.spring.TestKinesisProducerApplication;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.Message;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.web.servlet.MockMvc;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
 
 @ActiveProfiles({PROFILE_TEST})
-@SpringBootTest(webEnvironment = RANDOM_PORT, classes = TestKinesisProducerApplication.class)
+@SpringBootTest(
+        webEnvironment = RANDOM_PORT,
+        properties = {
+            "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.idleBetweenPolls = 1",
+            "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.listenerMode = batch",
+            "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.checkpointMode = manual",
+            "spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.useNativeDecoding = true",
+            "spring.cloud.stream.kinesis.binder.headers = event.eventType",
+            "spring.cloud.stream.kinesis.binder.autoAddShards = true"
+        },
+        classes = TestKinesisProducerApplication.class)
 @AutoConfigureMockMvc
 public abstract class AbstractIntegrationTest {
 
     @Autowired protected MockMvc mockMvc;
 
     @Autowired protected ObjectMapper objectMapper;
+
+    @Autowired protected KinesisAsyncClient amazonKinesis;
+
+    @Autowired protected CountDownLatch messageBarrier;
+
+    @Autowired protected AtomicReference<Message<List<?>>> messageHolder;
 }