juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/persistence/inmemory/ShardedChatHomeTest.java

   1 package de.juplo.kafka.chat.backend.persistence.inmemory;
   2
   3 import com.fasterxml.jackson.databind.ObjectMapper;
   4 import de.juplo.kafka.chat.backend.domain.ChatHomeWithShardsTestBase;
   5 import de.juplo.kafka.chat.backend.persistence.StorageStrategy;
   6 import de.juplo.kafka.chat.backend.persistence.storage.files.FilesStorageStrategy;
   7 import org.springframework.boot.test.context.TestConfiguration;
   8 import org.springframework.context.annotation.Bean;
   9
  10 import java.nio.file.Paths;
  11 import java.time.Clock;
  12 import java.util.stream.IntStream;
  13
  14
  15 public class ShardedChatHomeTest extends ChatHomeWithShardsTestBase
  16 {
  17   @TestConfiguration
  18   static class Configuration
  19   {
  20     @Bean
  21     ShardedChatHome chatHome(
  22         InMemoryChatHomeService chatHomeService)
  23     {
  24       SimpleChatHome[] chatHomes = new SimpleChatHome[numShards()];
  25
  26       IntStream
  27           .of(ownedShards())
  28           .forEach(shard -> chatHomes[shard] = new SimpleChatHome(chatHomeService, shard));
  29
  30       ShardingStrategy strategy = new KafkaLikeShardingStrategy(numShards());
  31
  32       return new ShardedChatHome(chatHomes, strategy);
  33     }
  34
  35     @Bean
  36     InMemoryChatHomeService chatHomeService(
  37         StorageStrategy storageStrategy)
  38     {
  39       return new InMemoryChatHomeService(
  40           numShards(),
  41           ownedShards(),
  42           storageStrategy.read());
  43     }
  44
  45     @Bean
  46     public FilesStorageStrategy storageStrategy()
  47     {
  48       return new FilesStorageStrategy(
  49           Paths.get("target", "test-classes", "data", "files"),
  50           Clock.systemDefaultZone(),
  51           8,
  52           new KafkaLikeShardingStrategy(numShards()),
  53           messageFlux -> new InMemoryChatRoomService(messageFlux),
  54           new ObjectMapper());
  55     }
  56
  57     Integer numShards()
  58     {
  59       return 10;
  60     }
  61
  62     int[] ownedShards()
  63     {
  64       return new int[] { 2 };
  65     }
  66   }
  67 }