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         Integer numShards,
  23         int[] ownedShards,
  24         InMemoryChatHomeService chatHomeService)
  25     {
  26       SimpleChatHome[] chatHomes = new SimpleChatHome[numShards];
  27
  28       IntStream
  29           .of(ownedShards)
  30           .forEach(shard -> chatHomes[shard] = new SimpleChatHome(chatHomeService, shard));
  31
  32       ShardingStrategy strategy = new KafkaLikeShardingStrategy(numShards);
  33
  34       return new ShardedChatHome(chatHomes, strategy);
  35     }
  36
  37     @Bean
  38     InMemoryChatHomeService chatHomeService(
  39         Integer numShards,
  40         int[] ownedShards,
  41         StorageStrategy storageStrategy)
  42     {
  43       return new InMemoryChatHomeService(
  44           numShards,
  45           ownedShards,
  46           storageStrategy.read());
  47     }
  48
  49     @Bean
  50     public FilesStorageStrategy storageStrategy(Integer numShards)
  51     {
  52       return new FilesStorageStrategy(
  53           Paths.get("target", "test-classes", "data", "files"),
  54           Clock.systemDefaultZone(),
  55           8,
  56           new KafkaLikeShardingStrategy(numShards),
  57           messageFlux -> new InMemoryChatRoomService(messageFlux),
  58           new ObjectMapper());
  59     }
  60
  61     @Bean
  62     Integer numShards()
  63     {
  64       return 10;
  65     }
  66
  67     @Bean
  68     int[] ownedShards()
  69     {
  70       return new int[] { 2 };
  71     }
  72   }
  73 }