stef_buwalda/pism
1
0
Fork 0
mirror of https://github.com/2OOP/pism.git synced 2026-02-04 10:54:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor to make Events easier to work with.

Browse Source
This commit is contained in:
lieght
2025-10-15 19:05:09 +02:00
parent bc6182e29a
commit 798005de2c
6 changed files with 78 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
framework/src/main/java/org/toop/framework/networking/NetworkingClientEventListener.java
View File

@@ -39,12 +39,13 @@ public class NetworkingClientEventListener {
}
void handleStartClient(NetworkEvents.StartClient event) {
long clientId = SnowflakeGenerator.nextId();
clientManager.startClient(
event.identifier(),
event.networkingClientClass(),
event.host(),
event.port(),
event.networkingReconnect()
clientId,
event.networkingClient(),
event.networkingConnector(),
() -> new EventFlow().addPostEvent(new NetworkEvents.StartClientResponse(clientId, true, event.identifier())).postEvent(),
() -> new EventFlow().addPostEvent(new NetworkEvents.StartClientResponse(clientId, false, event.identifier())).postEvent()
);
}
@@ -110,19 +111,23 @@ public class NetworkingClientEventListener {
}
private void handleReconnect(NetworkEvents.Reconnect event) {
try {
clientManager.reconnect(event.clientId(), event.networkingReconnect());
} catch (ClientNotFoundException e) {
logger.error(e);
}
clientManager.startClient(
event.clientId(),
event.networkingClient(),
event.networkingConnector(),
() -> new EventFlow().addPostEvent(new NetworkEvents.ReconnectResponse(true, event.identifier())).postEvent(),
() -> new EventFlow().addPostEvent(new NetworkEvents.ReconnectResponse(false, event.identifier())).postEvent()
);
}
private void handleChangeAddress(NetworkEvents.ChangeAddress event) {
try {
clientManager.changeAddress(event.clientId(), event.ip(), event.port(), event.networkingReconnect());
} catch (ClientNotFoundException e) {
logger.error(e);
}
clientManager.startClient(
event.clientId(),
event.networkingClient(),
event.networkingConnector(),
() -> new EventFlow().addPostEvent(new NetworkEvents.ChangeAddressResponse(true, event.identifier())).postEvent(),
() -> new EventFlow().addPostEvent(new NetworkEvents.ChangeAddressResponse(false, event.identifier())).postEvent()
);
}
void handleCloseClient(NetworkEvents.CloseClient event) {

153
framework/src/main/java/org/toop/framework/networking/NetworkingClientManager.java
View File

@@ -1,6 +1,5 @@
package org.toop.framework.networking;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
@@ -9,12 +8,10 @@ import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.toop.framework.eventbus.EventFlow;
import org.toop.framework.networking.events.NetworkEvents;
import org.toop.framework.networking.exceptions.ClientNotFoundException;
import org.toop.framework.networking.exceptions.CouldNotConnectException;
import org.toop.framework.networking.interfaces.NetworkingClient;
import org.toop.framework.networking.types.NetworkingReconnect;
import org.toop.framework.networking.types.NetworkingConnector;
public class NetworkingClientManager implements org.toop.framework.networking.interfaces.NetworkingClientManager {
private static final Logger logger = LogManager.getLogger(NetworkingClientManager.class);
@@ -22,13 +19,13 @@ public class NetworkingClientManager implements org.toop.framework.networking.in
public NetworkingClientManager() {}
@Override
public void startClient(
private void connectHelper(
long id,
NetworkingClient networkingClient,
String host, int port,
NetworkingReconnect networkingReconnect) {
NetworkingClient nClient,
NetworkingConnector nConnector,
Runnable onSuccess,
Runnable onFailure
) {
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
Runnable connectTask = new Runnable() {
@@ -36,26 +33,31 @@ public class NetworkingClientManager implements org.toop.framework.networking.in
@Override
public void run() {
if (networkClients.get(id) != null) {
networkClients.remove(id);
}
try {
networkingClient.connect(id, host, port);
networkClients.put(id, networkingClient);
logger.info("New client started successfully for {}:{}", host, port);
new EventFlow().addPostEvent(new NetworkEvents.StartClientResponse(id, false, id)).postEvent();
nClient.connect(id, nConnector.host(), nConnector.port());
networkClients.put(id, nClient);
logger.info("New client started successfully for {}:{}", nConnector.host(), nConnector.port());
onSuccess.run();
scheduler.shutdown();
} catch (CouldNotConnectException e) {
attempts++;
if (attempts < networkingReconnect.reconnectAttempts()) {
if (attempts < nConnector.reconnectAttempts()) {
logger.warn("Could not connect to {}:{}. Retrying in {} {}",
host, port, networkingReconnect.timeout(), networkingReconnect.timeUnit());
scheduler.schedule(this, networkingReconnect.timeout(), networkingReconnect.timeUnit());
nConnector.host(), nConnector.port(), nConnector.timeout(), nConnector.timeUnit());
scheduler.schedule(this, nConnector.timeout(), nConnector.timeUnit());
} else {
logger.error("Failed to start client for {}:{} after {} attempts", host, port, attempts);
new EventFlow().addPostEvent(new NetworkEvents.StartClientResponse(-1, false, id)).postEvent();
logger.error("Failed to start client for {}:{} after {} attempts", nConnector.host(), nConnector.port(), attempts);
onFailure.run();
scheduler.shutdown();
}
} catch (Exception e) {
logger.error("Unexpected exception during startClient", e);
new EventFlow().addPostEvent(new NetworkEvents.StartClientResponse(-1, false, id)).postEvent();
onFailure.run();
scheduler.shutdown();
}
}
@@ -64,6 +66,23 @@ public class NetworkingClientManager implements org.toop.framework.networking.in
scheduler.schedule(connectTask, 0, TimeUnit.MILLISECONDS);
}
@Override
public void startClient(
long id,
NetworkingClient nClient,
NetworkingConnector nConnector,
Runnable onSuccess,
Runnable onFailure
) {
connectHelper(
id,
nClient,
nConnector,
onSuccess,
onFailure
);
}
@Override
public void sendCommand(long id, String command) throws ClientNotFoundException {
logger.info("Sending command to client for {}:{}", id, command);
@@ -85,100 +104,6 @@ public class NetworkingClientManager implements org.toop.framework.networking.in
}
@Override
public void reconnect(long id, NetworkingReconnect networkingReconnect) throws ClientNotFoundException {
NetworkingClient client = this.networkClients.get(id);
if (client == null) {
throw new ClientNotFoundException(id);
}
InetSocketAddress address = client.getAddress();
if (client.isActive()) {
client.closeConnection();
}
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
Runnable connectTask = new Runnable() {
int attempts = 0;
@Override
public void run() {
try {
client.connect(id, address.getHostName(), address.getPort());
networkClients.put(id, client);
logger.info("Client {} reconnected to {}:{}", id, address.getHostName(), address.getPort());
new EventFlow().addPostEvent(new NetworkEvents.ReconnectResponse(true, id)).postEvent().postEvent();
scheduler.shutdown();
} catch (CouldNotConnectException e) {
attempts++;
if (attempts < networkingReconnect.reconnectAttempts()) {
logger.warn("Could not reconnect client {} to {}:{}. Retrying in {} {}",
id, address.getHostName(), address.getPort(), networkingReconnect.timeout(), networkingReconnect.timeUnit());
scheduler.schedule(this, networkingReconnect.timeout(), networkingReconnect.timeUnit());
} else {
logger.error("Failed to reconnect client {} to {}:{} after {} attempts", id, address.getHostName(), address.getPort(), attempts);
new EventFlow().addPostEvent(new NetworkEvents.ReconnectResponse(false, id)).postEvent().postEvent();
scheduler.shutdown();
}
} catch (Exception e) {
logger.error("Unexpected exception during reconnect for client {}", id, e);
new EventFlow().addPostEvent(new NetworkEvents.ReconnectResponse(false, id)).postEvent().postEvent();
scheduler.shutdown();
}
}
};
scheduler.schedule(connectTask, 0, TimeUnit.MILLISECONDS);
}
@Override
public void changeAddress(long id, String host, int port, NetworkingReconnect networkingReconnect) throws ClientNotFoundException {
NetworkingClient client = this.networkClients.get(id);
if (client == null) {
throw new ClientNotFoundException(id);
}
if (client.isActive()) {
client.closeConnection();
}
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
Runnable connectTask = new Runnable() {
int attempts = 0;
@Override
public void run() {
try {
client.connect(id, host, port);
networkClients.put(id, client);
logger.info("Client {} changed address to {}:{}", id, host, port);
new EventFlow().addPostEvent(new NetworkEvents.ChangeAddressResponse(true, id)).postEvent().postEvent();
scheduler.shutdown();
} catch (CouldNotConnectException e) {
attempts++;
if (attempts < networkingReconnect.reconnectAttempts()) {
logger.warn("Could not connect client {} to {}:{}. Retrying in {} {}",
id, host, port, networkingReconnect.timeout(), networkingReconnect.timeUnit());
scheduler.schedule(this, networkingReconnect.timeout(), networkingReconnect.timeUnit());
} else {
logger.error("Failed to connect client {} to {}:{} after {} attempts", id, host, port, attempts);
new EventFlow().addPostEvent(new NetworkEvents.ChangeAddressResponse(false, id)).postEvent().postEvent();
scheduler.shutdown();
}
} catch (Exception e) {
logger.error("Unexpected exception during changeAddress for client {}", id, e);
new EventFlow().addPostEvent(new NetworkEvents.ChangeAddressResponse(false, id)).postEvent().postEvent();
scheduler.shutdown();
}
}
};
scheduler.schedule(connectTask, 0, TimeUnit.MILLISECONDS);
}
@Override
public void closeClient(long id) throws ClientNotFoundException {
NetworkingClient client = this.networkClients.get(id);

22
framework/src/main/java/org/toop/framework/networking/events/NetworkEvents.java
View File

@@ -9,7 +9,7 @@ import org.toop.framework.eventbus.events.UniqueEvent;
import org.toop.framework.eventbus.events.EventsBase;
import org.toop.annotations.AutoResponseResult;
import org.toop.framework.networking.interfaces.NetworkingClient;
import org.toop.framework.networking.types.NetworkingReconnect;
import org.toop.framework.networking.types.NetworkingConnector;
/**
* A collection of networking-related event records for use with the {@link
@@ -118,16 +118,12 @@ public class NetworkEvents extends EventsBase {
*
* <p>Carries IP, port, and a unique event ID for correlation with responses.
*
* @param networkingClientClass The type of networking client to create.
* @param host Server IP address.
* @param port Server port.
* @param identifier Unique event identifier for correlation.
* @param networkingClient
* @param networkingConnector
*/
public record StartClient(
NetworkingClient networkingClientClass,
String host,
int port,
NetworkingReconnect networkingReconnect,
NetworkingClient networkingClient,
NetworkingConnector networkingConnector,
long identifier) implements UniqueEvent {}
/**
@@ -135,13 +131,12 @@ public class NetworkEvents extends EventsBase {
*
* @param clientId The client ID assigned to the new connection.
* @param successful If successfully connected or not. If not clientId will also be -1.
* @param identifier Event ID used for correlation.
*/
@AutoResponseResult
public record StartClientResponse(long clientId, boolean successful, long identifier) implements ResponseToUniqueEvent {}
/** WIP (Not working) Request to reconnect a client to a previous address. */
public record Reconnect(long clientId, NetworkingReconnect networkingReconnect, long identifier)
public record Reconnect(long clientId, NetworkingClient networkingClient, NetworkingConnector networkingConnector, long identifier)
implements UniqueEvent {}
public record ReconnectResponse(boolean successful, long identifier) implements ResponseToUniqueEvent {}
@@ -150,10 +145,9 @@ public class NetworkEvents extends EventsBase {
* Request to change a client connection to a new server.
*
* @param clientId The client connection ID.
* @param ip The new server IP.
* @param port The new server port.
* @param networkingConnector
*/
public record ChangeAddress(long clientId, String ip, int port, NetworkingReconnect networkingReconnect, long identifier)
public record ChangeAddress(long clientId, NetworkingClient networkingClient, NetworkingConnector networkingConnector, long identifier)
implements UniqueEvent {}
public record ChangeAddressResponse(boolean successful, long identifier) implements ResponseToUniqueEvent {}

15
framework/src/main/java/org/toop/framework/networking/interfaces/NetworkingClientManager.java
View File

@@ -2,19 +2,16 @@ package org.toop.framework.networking.interfaces;
import org.toop.framework.networking.exceptions.ClientNotFoundException;
import org.toop.framework.networking.exceptions.CouldNotConnectException;
import org.toop.framework.networking.types.NetworkingReconnect;
import java.util.Optional;
import org.toop.framework.networking.types.NetworkingConnector;
public interface NetworkingClientManager {
void startClient(
long id,
NetworkingClient networkingClientClass,
String host,
int port,
NetworkingReconnect networkingReconnect) throws CouldNotConnectException;
NetworkingClient nClient,
NetworkingConnector nConnector,
Runnable onSuccess,
Runnable onFailure
) throws CouldNotConnectException;
void sendCommand(long id, String command) throws ClientNotFoundException;
void reconnect(long id, NetworkingReconnect networkingReconnect) throws ClientNotFoundException;
void changeAddress(long id, String host, int port, NetworkingReconnect networkingReconnect) throws ClientNotFoundException;
void closeClient(long id) throws ClientNotFoundException;
}

5
framework/src/main/java/org/toop/framework/networking/types/NetworkingConnector.java Normal file
View File

@@ -0,0 +1,5 @@
package org.toop.framework.networking.types;
import java.util.concurrent.TimeUnit;
public record NetworkingConnector(String host, int port, int reconnectAttempts, long timeout, TimeUnit timeUnit) {}

5
framework/src/main/java/org/toop/framework/networking/types/NetworkingReconnect.java
View File

@@ -1,5 +0,0 @@
package org.toop.framework.networking.types;
import java.util.concurrent.TimeUnit;
public record NetworkingReconnect(int reconnectAttempts, long timeout, TimeUnit timeUnit) {}