scrow/scrow-monorepo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

.

Browse source
This commit is contained in:
kento2 2026-06-01 09:57:58 +02:00
parent aa6c8a57e3
commit cc402a71e7
4 changed files with 33 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

27
core-velocity/src/main/java/de/kentoj/scrow/corevelocity/ServerRegisterWatchdog.java
View file

@ -7,6 +7,7 @@ import de.kentoj.scrowlib.messaging.NatsRepository;
import de.kentoj.scrowlib.servermanager.ServerManager;
import io.nats.client.Connection;
import lombok.extern.java.Log;
import org.jetbrains.annotations.NotNull;
import reactor.core.publisher.Mono;
import java.net.InetSocketAddress;
@ -29,15 +30,27 @@ public class ServerRegisterWatchdog {
.map(DocumentRepository::fromMessage)
.flatMap(doc -> {
var state = doc.getString("state");
if (!"UP".equals(state)) return Mono.empty();
var handle = doc.getString("handle");
return serverManager.getPort(handle)
.map(port -> {
log.info("registering " + handle);
return server.registerServer(new ServerInfo(handle, new InetSocketAddress(port)));
});
return handleStateChange(state, handle);
})
.subscribe();
}
private @NotNull Mono<@NotNull Void> handleStateChange(String state, String handle) {
return switch (state) {
case "UP" -> serverManager.getInfo(handle)
.doOnNext(instance -> {
log.info("registering " + handle);
server.registerServer(new ServerInfo(handle, new InetSocketAddress(instance.getPort())));
})
.then();
case "DOWN" -> serverManager.getInfo(handle)
.doOnNext(instance -> {
log.info("unregistering " + handle);
server.unregisterServer(new ServerInfo(handle, new InetSocketAddress(instance.getPort())));
})
.then();
default -> Mono.empty();
};
}
}