Integration tests with Kafka

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/DummyService.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/DummyService.java
new file mode 100644 (file)
index 0000000..70f7a62
--- /dev/null
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/DummyService.java
@@ -0,0 +1,16 @@
+package de.example.spring.kafka;
+
+import javax.inject.Named;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Named
+public class DummyService {
+       private static final Logger LOGGER = LoggerFactory.getLogger(DummyService.class);
+
+
+       public void iAmVeryDummy(String message) {
+               LOGGER.info("I am a dummy service: '{}'", message);
+       }
+}

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/Receiver.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/Receiver.java
index 7c88cd5..7681ab5 100644 (file)
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/Receiver.java
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/Receiver.java
@@ -1,5 +1,7 @@
 package de.example.spring.kafka;
 
+import javax.inject.Inject;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.cloud.stream.annotation.EnableBinding;
@@ -9,11 +11,20 @@ import org.springframework.cloud.stream.messaging.Sink;
 @EnableBinding(Sink.class)
 public class Receiver {
   private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);
+  
+  private final DummyService dummyService;
+
+  @Inject
+  public Receiver(DummyService dummyService) {
+       this.dummyService = dummyService;
+  }
 
   @StreamListener(Sink.INPUT)
   public void handle(Product product) {
     LOGGER.info("product name='{}'", product.getName());
     LOGGER.info("product description='{}'", product.getDescription());
+    
+    dummyService.iAmVeryDummy(product.getName());
   }
 
 }

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java
index e385776..f635538 100644 (file)
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java
@@ -7,8 +7,8 @@ import org.springframework.context.annotation.Configuration;
 public class ReceiverConfig {
 
   @Bean
-  public Receiver receiver() {
-    return new Receiver();
+  public Receiver receiver(DummyService dummyService) {
+    return new Receiver(dummyService);
   }
 
 

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/test/java/de/example/spring/kafka/ReceiverShould.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/test/java/de/example/spring/kafka/ReceiverShould.java
new file mode 100644 (file)
index 0000000..6291ea4
--- /dev/null
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/test/java/de/example/spring/kafka/ReceiverShould.java
@@ -0,0 +1,45 @@
+package de.example.spring.kafka;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.doNothing;
+
+import javax.inject.Inject;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = { Receiver.class })
+@DirtiesContext
+public class ReceiverShould {
+
+       @Inject
+       Sink source;
+       
+       @MockBean
+       DummyService dummyService;
+       
+       @Test
+       public void callSomeDummy() {
+               String productName = "product";
+               String productDescription = "productDescription";
+               Product product = new Product(productName, productDescription);
+           ArgumentCaptor<String> dummyArgCaptor = ArgumentCaptor.forClass(String.class);
+               doNothing().when(dummyService).iAmVeryDummy(dummyArgCaptor.capture());
+               
+           Message<Product> message = new GenericMessage<>(product);
+           source.input().send(message);
+
+           assertThat(dummyArgCaptor.getValue(), is(product.getName()));
+       }
+
+}

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/test/java/de/example/spring/kafka/SenderShould.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/test/java/de/example/spring/kafka/SenderShould.java
new file mode 100644 (file)
index 0000000..71fbfec
--- /dev/null
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/test/java/de/example/spring/kafka/SenderShould.java
@@ -0,0 +1,42 @@
+package de.example.spring.kafka;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import javax.inject.Inject;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = { Sender.class })
+@DirtiesContext
+public class SenderShould {
+
+       @Inject
+       Source source;
+       
+       @Inject
+       Sender sender;
+       
+       @Inject
+       private MessageCollector messageCollector;
+       
+       @Test
+       public void sendSomeProduct() {
+               Product product = new Product("hello", "this is some description");
+               
+               sender.sendMessage("hello");
+               
+               Message<Product> received = (Message<Product>) messageCollector.forChannel(source.output()).poll();
+               
+           assertThat(received.getPayload().getDescription(), is(product.getDescription()));
+       }
+
+}