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.ChatHomeWithShardsTest;
   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 public class ShardedChatHomeTest extends ChatHomeWithShardsTest
  15 {
  16   @TestConfiguration
  17   static class Configuration
  18   {
  19     @Bean
  20     ShardedChatHome chatHome(
  21         StorageStrategy storageStrategy,
  22         Clock clock)
  23     {
  24       SimpleChatHome[] chatHomes = new SimpleChatHome[NUM_SHARDS];
  25
  26       IntStream
  27           .of(ownedShards())
  28           .forEach(shard -> chatHomes[shard] = new SimpleChatHome(
  29               shard,
  30               storageStrategy.read(),
  31               clock,
  32               bufferSize()));
  33
  34       ShardingStrategy strategy = new KafkaLikeShardingStrategy(NUM_SHARDS);
  35
  36       return new ShardedChatHome(chatHomes, strategy);
  37     }
  38
  39     @Bean
  40     public FilesStorageStrategy storageStrategy(Clock clock)
  41     {
  42       return new FilesStorageStrategy(
  43           Paths.get("target", "test-classes", "data", "files"),
  44           clock,
  45           bufferSize(),
  46           new KafkaLikeShardingStrategy(NUM_SHARDS),
  47           messageFlux -> new InMemoryChatRoomService(messageFlux),
  48           new ObjectMapper());
  49     }
  50
  51     @Bean
  52     Clock clock()
  53     {
  54       return Clock.systemDefaultZone();
  55     }
  56
  57     int[] ownedShards()
  58     {
  59       return new int[] { OWNED_SHARD };
  60     }
  61
  62     int bufferSize()
  63     {
  64       return 8;
  65     }
  66   }
  67 }