- log.debug("Creating InMemoryChatHomeService with buffer-size {} (for created ChatRoom's)", bufferSize);
- this.chatrooms = new HashMap<>();
- chatroomFlux.toStream().forEach(chatroom -> chatrooms.put(chatroom.getId(), chatroom));
- this.clock = clock;
- this.bufferSize = bufferSize;
+ log.debug("Creating InMemoryChatHomeService");
+ this.chatrooms = new Map[numShards];
+ Set<Integer> owned = Arrays
+ .stream(ownedShards)
+ .collect(
+ () -> new HashSet<>(),
+ (set, i) -> set.add(i),
+ (a, b) -> a.addAll(b));
+ for (int shard = 0; shard < numShards; shard++)
+ {
+ chatrooms[shard] = owned.contains(shard)
+ ? new HashMap<>()
+ : null;
+ }
+ chatroomFlux
+ .filter(chatRoom ->
+ {
+ if (owned.contains(chatRoom.getShard()))
+ {
+ return true;
+ }
+ else
+ {
+ log.info("Ignoring not owned chat-room {}", chatRoom);
+ return false;
+ }
+ })
+ .toStream()
+ .forEach(chatroom -> chatrooms[chatroom.getShard()].put(chatroom.getId(), chatroom));