Dragonestia/RoomPicker
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added linking between nodes, buckets, users

Browse Source
This commit is contained in:
Andrey Terentev 2023-11-16 13:01:02 +07:00
parent ab446e7160
commit 8a14640aaa
4 changed files with 37 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/ru/dragonestia/loadbalancer/repository/UserRepository.java
View File

@ -5,16 +5,15 @@ import ru.dragonestia.loadbalancer.model.User;
import java.util.Collection;
import java.util.List;
import java.util.Map;
public interface UserRepository {
void linkWithBucket(Bucket bucket, Collection<User> users);
Map<User, Boolean> linkWithBucket(Bucket bucket, Collection<User> users);
void unlinkWithBucket(Bucket bucket, Collection<User> users);
int unlinkWithBucket(Bucket bucket, Collection<User> users);
List<Bucket> findAllLinkedUserBuckets(User user);
int tryUnlinkWithBucket(Bucket bucket, Collection<User> users);
void onRemoveBucket(Bucket bucket);
}

12
src/main/java/ru/dragonestia/loadbalancer/repository/impl/BucketRepositoryImpl.java
View File

@ -49,6 +49,8 @@ public class BucketRepositoryImpl implements BucketRepository {
node2bucketsMap.get(node.get()).remove(bucket.getIdentifier());
}
userRepository.onRemoveBucket(bucket);
}
@Override
@ -85,10 +87,16 @@ public class BucketRepositoryImpl implements BucketRepository {
}
var requiredSlots = users.size();
var container = node2bucketsMap.get(node).values().stream()
var container = node2bucketsMap.get(node).values().stream() // TODO: pick bucket with used node balancing method
.filter(b -> b.isAvailable(requiredSlots))
.findFirst();
if (container.isPresent()) {
var cont = container.get();
var addedUsers = userRepository.linkWithBucket(cont.bucket(), users);
cont.used().getAndAdd((int) addedUsers.values().stream().filter(Boolean.TRUE::equals).count());
}
return container.map(BucketContainer::bucket);
}
}
@ -110,7 +118,7 @@ public class BucketRepositoryImpl implements BucketRepository {
throw new IllegalArgumentException("Bucket '" + nodeId + "' does not exist");
}
var delta = userRepository.tryUnlinkWithBucket(bucket, users);
var delta = userRepository.unlinkWithBucket(bucket, users);
if (buckets.get(bucket.getIdentifier()).used().getAndAdd(-delta) < 0) {
throw new RuntimeException("Bucket has less than 0 users");
}

10
src/main/java/ru/dragonestia/loadbalancer/repository/impl/NodeRepositoryImpl.java
View File

@ -1,6 +1,9 @@
package ru.dragonestia.loadbalancer.repository.impl;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Repository;
import ru.dragonestia.loadbalancer.model.Node;
import ru.dragonestia.loadbalancer.repository.BucketRepository;
import ru.dragonestia.loadbalancer.repository.NodeRepository;
import java.util.List;
@ -8,8 +11,11 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@Repository
@RequiredArgsConstructor
public class NodeRepositoryImpl implements NodeRepository {
private final BucketRepository bucketRepository;
private final Map<String, Node> nodeMap = new ConcurrentHashMap<>();
@Override
@ -21,6 +27,8 @@ public class NodeRepositoryImpl implements NodeRepository {
nodeMap.put(node.identifier(), node);
}
bucketRepository.onCreateNode(node);
}
@Override
@ -28,6 +36,8 @@ public class NodeRepositoryImpl implements NodeRepository {
synchronized (nodeMap) {
nodeMap.remove(node.identifier());
}
bucketRepository.onRemoveNode(node);
}
@Override

40
src/main/java/ru/dragonestia/loadbalancer/repository/impl/UserRepositoryImpl.java
View File

@ -15,41 +15,22 @@ public class UserRepositoryImpl implements UserRepository {
private final Map<User, Set<Bucket>> usersMap = new ConcurrentHashMap<>();
@Override
public void linkWithBucket(Bucket bucket, Collection<User> users) {
public Map<User, Boolean> linkWithBucket(Bucket bucket, Collection<User> users) {
var result = new HashMap<User, Boolean>();
synchronized (usersMap) {
for (var user: users) {
var set = usersMap.getOrDefault(user, new HashSet<>());
set.add(bucket);
result.put(user, set.add(bucket));
usersMap.put(user, set);
}
}
return result;
}
@Override
public void unlinkWithBucket(Bucket bucket, Collection<User> users) {
synchronized (usersMap) {
for (var user: users) {
var set = usersMap.getOrDefault(user, new HashSet<>());
set.remove(bucket);
if (set.isEmpty()) {
usersMap.remove(user);
} else {
usersMap.put(user, set);
}
}
}
}
@Override
public List<Bucket> findAllLinkedUserBuckets(User user) {
synchronized (usersMap) {
return usersMap.getOrDefault(user, new HashSet<>()).stream().toList();
}
}
@Override
public int tryUnlinkWithBucket(Bucket bucket, Collection<User> users) {
public int unlinkWithBucket(Bucket bucket, Collection<User> users) {
var counter = new AtomicInteger();
synchronized (usersMap) {
usersMap.forEach((user, set) -> {
@ -66,6 +47,13 @@ public class UserRepositoryImpl implements UserRepository {
return counter.get();
}
@Override
public List<Bucket> findAllLinkedUserBuckets(User user) {
synchronized (usersMap) {
return usersMap.getOrDefault(user, new HashSet<>()).stream().toList();
}
}
@Override
public void onRemoveBucket(Bucket bucket) {
synchronized (usersMap) {