1 package de.juplo.kafka.chat.backend.implementation.inmemory;
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;
12 import java.util.Arrays;
13 import java.util.HashSet;
15 import java.util.UUID;
16 import java.util.stream.Collectors;
20 public class ShardedChatHomeService implements ChatHomeService
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;
29 public ShardedChatHomeService(
31 SimpleChatHomeService[] chatHomes,
33 ShardingStrategy shardingStrategy)
35 this.instanceId = instanceId;
36 this.chatHomes = chatHomes;
37 this.shardOwners = Arrays
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);
47 "Created ShardedChatHome for shards: {}",
51 .collect(Collectors.joining(", ")));
56 public Mono<ChatRoomInfo> createChatRoom(UUID id, String name)
58 int shard = shardingStrategy.selectShard(id);
59 return chatHomes[shard] == null
60 ? Mono.error(new ShardNotOwnedException(instanceId, shard))
61 : chatHomes[shard].createChatRoom(id, name);
65 public Mono<ChatRoomInfo> getChatRoomInfo(UUID id)
67 int shard = selectShard(id);
68 return chatHomes[shard] == null
69 ? Mono.error(new ShardNotOwnedException(instanceId, shard))
72 .onErrorMap(throwable -> throwable instanceof UnknownChatroomException
73 ? new UnknownChatroomException(
76 ownedShards.stream().mapToInt(i -> i.intValue()).toArray())
81 public Flux<ChatRoomInfo> getChatRoomInfo()
84 .fromIterable(ownedShards)
85 .flatMap(shard -> chatHomes[shard].getChatRoomInfo());
89 public Mono<ChatRoomData> getChatRoomData(UUID id)
91 int shard = selectShard(id);
92 return chatHomes[shard] == null
93 ? Mono.error(new ShardNotOwnedException(instanceId, shard))
96 .onErrorMap(throwable -> throwable instanceof UnknownChatroomException
97 ? new UnknownChatroomException(
100 ownedShards.stream().mapToInt(i -> i.intValue()).toArray())
105 public Mono<String[]> getShardOwners()
107 return Mono.just(shardOwners);
110 private int selectShard(UUID chatroomId)
112 return shardingStrategy.selectShard(chatroomId);