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.domain.ChatHome;
   5 import de.juplo.kafka.chat.backend.persistence.StorageStrategy;
   6 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
   7 import org.springframework.context.annotation.Bean;
   8 import org.springframework.context.annotation.Configuration;
   9
  10 import java.time.Clock;
  11 import java.util.stream.IntStream;
  12
  13
  14 @ConditionalOnProperty(
  15     prefix = "chat.backend",
  16     name = "services",
  17     havingValue = "inmemory",
  18     matchIfMissing = true)
  19 @Configuration
  20 public class InMemoryServicesConfiguration
  21 {
  22   @Bean
  23   @ConditionalOnProperty(
  24       prefix = "chat.backend.inmemory",
  25       name = "sharding-strategy",
  26       havingValue = "none",
  27       matchIfMissing = true)
  28   ChatHome noneShardingChatHome(
  29       ChatBackendProperties properties,
  30       StorageStrategy storageStrategy,
  31       Clock clock)
  32   {
  33     return new SimpleChatHome(
  34         storageStrategy,
  35         clock,
  36         properties.getChatroomBufferSize());
  37   }
  38
  39   @Bean
  40   @ConditionalOnProperty(
  41       prefix = "chat.backend.inmemory",
  42       name = "sharding-strategy",
  43       havingValue = "kafkalike")
  44   ChatHome kafkalikeShardingChatHome(
  45       ChatBackendProperties properties,
  46       StorageStrategy storageStrategy,
  47       Clock clock)
  48   {
  49     int numShards = properties.getInmemory().getNumShards();
  50     SimpleChatHome[] chatHomes = new SimpleChatHome[numShards];
  51     IntStream
  52         .of(properties.getInmemory().getOwnedShards())
  53         .forEach(shard -> chatHomes[shard] = new SimpleChatHome(
  54             shard,
  55             storageStrategy,
  56             clock,
  57             properties.getChatroomBufferSize()));
  58     ShardingStrategy strategy = new KafkaLikeShardingStrategy(numShards);
  59     return new ShardedChatHome(chatHomes, strategy);
  60   }
  61
  62   @ConditionalOnProperty(
  63       prefix = "chat.backend.inmemory",
  64       name = "sharding-strategy",
  65       havingValue = "none",
  66       matchIfMissing = true)
  67   @Bean
  68   ShardingStrategy defaultShardingStrategy()
  69   {
  70     return chatRoomId -> 0;
  71   }
  72
  73   @ConditionalOnProperty(
  74       prefix = "chat.backend.inmemory",
  75       name = "sharding-strategy",
  76       havingValue = "kafkalike")
  77   @Bean
  78   ShardingStrategy kafkalikeShardingStrategy(ChatBackendProperties properties)
  79   {
  80     return new KafkaLikeShardingStrategy(
  81         properties.getInmemory().getNumShards());
  82   }
  83 }