X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;fp=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;h=1336050a436a4990c9b8391faa06e86100326ce8;hb=ecadcd52ab7fe117fef3450a27b9d6ecdf621716;hp=1a336d684992afaf659fae096445456f67599704;hpb=55d4ad6e7ccbd5d284d7bcda7cfc857555fcd0a5;p=demos%2Fkafka%2Ftraining

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 1a336d6..1336050 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -3,21 +3,33 @@ package de.juplo.kafka;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.Bytes;
+import org.springframework.beans.factory.annotation.Autowired;
 
+import java.util.*;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 
 public class ApplicationTests extends GenericApplicationTests<String, String>
 {
+  @Autowired
+  AdderResults results;
+
+
   public ApplicationTests()
   {
     super(new ApplicationTestRecrodGenerator());
+    ((ApplicationTestRecrodGenerator)recordGenerator).tests = this;
   }
 
 
   static class ApplicationTestRecrodGenerator implements RecordGenerator
   {
+    ApplicationTests tests;
+
     final int[] numbers = {1, 7, 3, 2, 33, 6, 11};
     final String[] dieWilden13 =
         IntStream
@@ -29,6 +41,8 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
 
     int counter = 0;
 
+    Map<String, List<AdderResult>> state;
+
     @Override
     public int generate(
         boolean poisonPills,
@@ -36,6 +50,12 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
         Consumer<ProducerRecord<Bytes, Bytes>> messageSender)
     {
       counter = 0;
+      state =
+          Arrays
+              .stream(dieWilden13)
+              .collect(Collectors.toMap(
+                  seeräuber -> seeräuber,
+                  seeräuber -> new LinkedList()));
 
       for (int i = 0; i < 33; i++)
       {
@@ -50,6 +70,8 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
           send(key, value, logicErrors, messageSender);
         }
         send(key, calculateMessage, logicErrors, messageSender);
+
+        state.get(seeräuber).add(new AdderResult(number, (number + 1) * number / 2));
       }
 
       return counter;
@@ -79,5 +101,24 @@ public class ApplicationTests extends GenericApplicationTests<String, String>
     {
       return false;
     }
+
+    @Override
+    public void assertBusinessLogic()
+    {
+      tests.results
+          .getState()
+          .values()
+          .stream()
+          .flatMap(map -> map.entrySet().stream())
+          .forEach(entry ->
+          {
+            String user = entry.getKey();
+            List<AdderResult> resultsForUser = entry.getValue();
+
+            assertThat(state.get(user))
+                .describedAs("Unexpected results for user {}", user)
+                .containsExactlyElementsOf(resultsForUser);
+          });
+    }
   }
 }