juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/persistence/inmemory/InMemoryServicesConfiguration.java

   1 package de.juplo.kafka.chat.backend.persistence.inmemory;
   2
   3 import de.juplo.kafka.chat.backend.ChatBackendProperties;
   4 import de.juplo.kafka.chat.backend.ChatBackendProperties.ShardingStrategyType;
   5 import de.juplo.kafka.chat.backend.domain.ShardedChatHome;
   6 import de.juplo.kafka.chat.backend.persistence.KafkaLikeShardingStrategy;
   7 import de.juplo.kafka.chat.backend.domain.ShardingStrategy;
   8 import de.juplo.kafka.chat.backend.domain.ChatHome;
   9 import de.juplo.kafka.chat.backend.domain.SimpleChatHome;
  10 import de.juplo.kafka.chat.backend.persistence.StorageStrategy;
  11 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  12 import org.springframework.context.annotation.Bean;
  13 import org.springframework.context.annotation.Configuration;
  14
  15 import java.time.Clock;
  16 import java.util.stream.IntStream;
  17
  18
  19 @ConditionalOnProperty(
  20     prefix = "chat.backend",
  21     name = "services",
  22     havingValue = "inmemory",
  23     matchIfMissing = true)
  24 @Configuration
  25 public class InMemoryServicesConfiguration
  26 {
  27   @Bean
  28   @ConditionalOnProperty(
  29       prefix = "chat.backend.inmemory",
  30       name = "sharding-strategy",
  31       havingValue = "none",
  32       matchIfMissing = true)
  33   ChatHome noneShardingChatHome(InMemoryChatHomeService chatHomeService)
  34   {
  35     return new SimpleChatHome(chatHomeService);
  36   }
  37
  38   @Bean
  39   @ConditionalOnProperty(
  40       prefix = "chat.backend.inmemory",
  41       name = "sharding-strategy",
  42       havingValue = "kafkalike")
  43   ChatHome kafkalikeShardingChatHome(
  44       ChatBackendProperties properties,
  45       InMemoryChatHomeService chatHomeService)
  46   {
  47     int numShards = properties.getInmemory().getNumShards();
  48     SimpleChatHome[] chatHomes = new SimpleChatHome[numShards];
  49     IntStream
  50         .of(properties.getInmemory().getOwnedShards())
  51         .forEach(shard -> chatHomes[shard] = new SimpleChatHome(chatHomeService, shard));
  52     ShardingStrategy strategy = new KafkaLikeShardingStrategy(numShards);
  53     return new ShardedChatHome(chatHomes, strategy);
  54   }
  55
  56   @Bean
  57   InMemoryChatHomeService chatHomeService(
  58       ChatBackendProperties properties,
  59       StorageStrategy storageStrategy)
  60   {
  61     ShardingStrategyType sharding =
  62         properties.getInmemory().getShardingStrategy();
  63     int numShards = sharding == ShardingStrategyType.none
  64         ? 1
  65         : properties.getInmemory().getNumShards();
  66     int[] ownedShards = sharding == ShardingStrategyType.none
  67         ? new int[] { 0 }
  68         : properties.getInmemory().getOwnedShards();
  69     return new InMemoryChatHomeService(
  70         numShards,
  71         ownedShards,
  72         storageStrategy.read());
  73   }
  74
  75   @Bean
  76   InMemoryChatRoomFactory chatRoomFactory(
  77       InMemoryChatHomeService service,
  78       ShardingStrategy strategy,
  79       Clock clock,
  80       ChatBackendProperties properties)
  81   {
  82     return new InMemoryChatRoomFactory(
  83         service,
  84         strategy,
  85         clock,
  86         properties.getChatroomBufferSize());
  87   }
  88
  89   @ConditionalOnProperty(
  90       prefix = "chat.backend.inmemory",
  91       name = "sharding-strategy",
  92       havingValue = "none",
  93       matchIfMissing = true)
  94   @Bean
  95   ShardingStrategy defaultShardingStrategy()
  96   {
  97     return chatRoomId -> 0;
  98   }
  99
 100   @ConditionalOnProperty(
 101       prefix = "chat.backend.inmemory",
 102       name = "sharding-strategy",
 103       havingValue = "kafkalike")
 104   @Bean
 105   ShardingStrategy kafkalikeShardingStrategy(ChatBackendProperties properties)
 106   {
 107     return new KafkaLikeShardingStrategy(
 108         properties.getInmemory().getNumShards());
 109   }
 110 }