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