juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/query/QueryStreamProcessorTopologyTest.java

   1 package de.juplo.kafka.wordcount.query;
   2
   3 import de.juplo.kafka.wordcount.top10.TestRanking;
   4 import de.juplo.kafka.wordcount.users.TestUserData;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.apache.kafka.common.serialization.StringSerializer;
   7 import org.apache.kafka.streams.TestInputTopic;
   8 import org.apache.kafka.streams.Topology;
   9 import org.apache.kafka.streams.TopologyTestDriver;
  10 import org.apache.kafka.streams.state.KeyValueStore;
  11 import org.apache.kafka.streams.state.Stores;
  12 import org.junit.jupiter.api.AfterEach;
  13 import org.junit.jupiter.api.BeforeEach;
  14 import org.junit.jupiter.api.Test;
  15 import org.springframework.kafka.support.serializer.JsonSerializer;
  16
  17 import java.util.Map;
  18
  19 import static de.juplo.kafka.wordcount.query.QueryApplicationConfiguration.serializationConfig;
  20
  21
  22 @Slf4j
  23 public class QueryStreamProcessorTopologyTest
  24 {
  25   public static final String TOP10_IN = "TOP10-IN";
  26   public static final String USERS_IN = "USERS-IN";
  27   public static final String STORE_NAME = "TOPOLOGY-TEST";
  28
  29
  30   TopologyTestDriver testDriver;
  31   TestInputTopic<String, TestRanking> top10In;
  32   TestInputTopic<String, TestUserData> userIn;
  33
  34
  35   @BeforeEach
  36   public void setUp()
  37   {
  38     Topology topology = QueryStreamProcessor.buildTopology(
  39         USERS_IN,
  40         TOP10_IN,
  41         Stores.inMemoryKeyValueStore(STORE_NAME));
  42
  43     testDriver = new TopologyTestDriver(topology, serializationConfig());
  44
  45     top10In = testDriver.createInputTopic(
  46         TOP10_IN,
  47         new StringSerializer(),
  48         jsonSerializer(TestRanking.class));
  49
  50     userIn = testDriver.createInputTopic(
  51         USERS_IN,
  52         new StringSerializer(),
  53         jsonSerializer(TestUserData.class));
  54   }
  55
  56
  57   @Test
  58   public void test()
  59   {
  60     TestData
  61         .getUsersMessages()
  62         .forEach(kv -> userIn.pipeInput(kv.key, kv.value));
  63     TestData
  64         .getTop10Messages()
  65         .forEach(kv -> top10In.pipeInput(kv.key, kv.value));
  66
  67     KeyValueStore<String, UserRanking> store = testDriver.getKeyValueStore(STORE_NAME);
  68     TestData.assertExpectedState(user -> store.get(user));
  69   }
  70
  71   @AfterEach
  72   public void tearDown()
  73   {
  74     testDriver.close();
  75   }
  76
  77   private <T> JsonSerializer<T> jsonSerializer(Class<T> type)
  78   {
  79     JsonSerializer<T> jsonSerializer = new JsonSerializer<>();
  80     jsonSerializer.configure(
  81         Map.of(
  82             JsonSerializer.TYPE_MAPPINGS,
  83             "userdata:" + TestUserData.class.getName() + "," +
  84             "ranking:" + TestRanking.class.getName()),
  85         false);
  86     return jsonSerializer;
  87   }
  88 }