juplo.de Git - demos/kafka/chat/blob

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