juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/ShardedChatHomeService.java

   1 package de.juplo.kafka.chat.backend.implementation.inmemory;
   2
   3 import de.juplo.kafka.chat.backend.domain.*;
   4 import de.juplo.kafka.chat.backend.domain.exceptions.ShardNotOwnedException;
   5 import de.juplo.kafka.chat.backend.domain.exceptions.UnknownChatroomException;
   6 import de.juplo.kafka.chat.backend.implementation.ShardingStrategy;
   7 import lombok.extern.slf4j.Slf4j;
   8 import reactor.core.publisher.Flux;
   9 import reactor.core.publisher.Mono;
  10
  11 import java.net.URI;
  12 import java.util.Arrays;
  13 import java.util.HashSet;
  14 import java.util.Set;
  15 import java.util.UUID;
  16 import java.util.stream.Collectors;
  17
  18
  19 @Slf4j
  20 public class ShardedChatHomeService implements ChatHomeService
  21 {
  22   private final String instanceId;
  23   private final SimpleChatHomeService[] chatHomes;
  24   private final Set<Integer> ownedShards;
  25   private final String[] shardOwners;
  26   private final ShardingStrategy shardingStrategy;
  27
  28
  29   public ShardedChatHomeService(
  30       String instanceId,
  31       SimpleChatHomeService[] chatHomes,
  32       URI[] shardOwners,
  33       ShardingStrategy shardingStrategy)
  34   {
  35     this.instanceId = instanceId;
  36     this.chatHomes = chatHomes;
  37     this.shardOwners = Arrays
  38         .stream(shardOwners)
  39         .map(uri -> uri.toASCIIString())
  40         .toArray(size -> new String[size]);
  41     this.shardingStrategy = shardingStrategy;
  42     this.ownedShards = new HashSet<>();
  43     for (int shard = 0; shard < chatHomes.length; shard++)
  44       if(chatHomes[shard] != null)
  45         this.ownedShards.add(shard);
  46     log.info(
  47         "Created ShardedChatHome for shards: {}",
  48         ownedShards
  49             .stream()
  50             .map(String::valueOf)
  51             .collect(Collectors.joining(", ")));
  52   }
  53
  54
  55   @Override
  56   public Mono<ChatRoomInfo> createChatRoom(UUID id, String name)
  57   {
  58     int shard = shardingStrategy.selectShard(id);
  59     return chatHomes[shard] == null
  60         ? Mono.error(new ShardNotOwnedException(instanceId, shard))
  61         : chatHomes[shard].createChatRoom(id, name);
  62   }
  63
  64   @Override
  65   public Mono<ChatRoomInfo> getChatRoomInfo(UUID id)
  66   {
  67     int shard = selectShard(id);
  68     return chatHomes[shard] == null
  69         ? Mono.error(new ShardNotOwnedException(instanceId, shard))
  70         : chatHomes[shard]
  71             .getChatRoomInfo(id)
  72             .onErrorMap(throwable -> throwable instanceof UnknownChatroomException
  73             ? new UnknownChatroomException(
  74                 id,
  75                 shard,
  76                 ownedShards.stream().mapToInt(i -> i.intValue()).toArray())
  77             : throwable);
  78   }
  79
  80   @Override
  81   public Flux<ChatRoomInfo> getChatRoomInfo()
  82   {
  83     return Flux
  84         .fromIterable(ownedShards)
  85         .flatMap(shard -> chatHomes[shard].getChatRoomInfo());
  86   }
  87
  88   @Override
  89   public Mono<ChatRoomData> getChatRoomData(UUID id)
  90   {
  91     int shard = selectShard(id);
  92     return chatHomes[shard] == null
  93         ? Mono.error(new ShardNotOwnedException(instanceId, shard))
  94         : chatHomes[shard]
  95             .getChatRoomData(id)
  96             .onErrorMap(throwable -> throwable instanceof UnknownChatroomException
  97                 ? new UnknownChatroomException(
  98                 id,
  99                 shard,
 100                 ownedShards.stream().mapToInt(i -> i.intValue()).toArray())
 101                 : throwable);
 102   }
 103
 104   @Override
 105   public Mono<String[]> getShardOwners()
 106   {
 107     return Mono.just(shardOwners);
 108   }
 109
 110   private int selectShard(UUID chatroomId)
 111   {
 112     return shardingStrategy.selectShard(chatroomId);
 113   }
 114 }