Add SessionReader and SessionWriter worker threads for session #198

Merged
lars.winzer merged 20 commits from feat/session-reader-and-writer-thread into main 2026-03-20 17:39:44 +01:00
4 changed files with 20 additions and 15 deletions
Showing only changes of commit b19e923356 - Show all commits
@@ -27,8 +27,7 @@ public class ServerApp {
EventBus eventBus = new EventBus();
SessionManager sessionManager = new SessionManager(eventBus);
eventBus.subscribe(
DisconnectEvent.class, event -> sessionManager.onDisconnect(event));
eventBus.subscribe(DisconnectEvent.class, event -> sessionManager.onDisconnect(event));
NetworkManager networkManager = new NetworkManager(port, sessionManager);
UserRegistry userRegistry = new UserRegistry();
@@ -13,6 +13,8 @@ public class Session {
private final TransportLayer transport;
private final BlockingQueue<PrimitiveResponse> responseQueue;
private final int RESPOND_QUEUE_SIZE = 32;
/**
* Creates a new Session with the given transport and event bus.
*
@@ -23,7 +25,7 @@ public class Session {
public Session(TransportLayer transport, EventBus eventBus) {
this.id = new SessionId();
this.transport = transport;
this.responseQueue = new ArrayBlockingQueue<>(32);
this.responseQueue = new ArrayBlockingQueue<>(RESPOND_QUEUE_SIZE);
}
/**
@@ -1,15 +1,14 @@
package ch.unibas.dmi.dbis.cs108.casono.server.network.sessions;
import ch.unibas.dmi.dbis.cs108.casono.server.network.events.DisconnectEvent;
import ch.unibas.dmi.dbis.cs108.casono.server.network.events.EventBus;
import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.TransportLayer;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import ch.unibas.dmi.dbis.cs108.casono.server.network.events.DisconnectEvent;
import ch.unibas.dmi.dbis.cs108.casono.server.network.events.EventBus;
import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.TransportLayer;
/** Manages active sessions in the server. */
public class SessionManager {
private Map<SessionId, SessionHandle> sessions;
@@ -25,9 +24,9 @@ public class SessionManager {
/**
* Create new Session from provided transport.
*
* <p> Will create both worker threads and start them.
*
*
* <p>Will create both worker threads and start them.
*
* @param transport to create session from
* @return newly created session
*/
@@ -50,15 +49,17 @@ public class SessionManager {
/**
* Disconnect specified client
*
* <p> WARNING: Client will be uninformed about disconnect. Use with caution.
*
*
* <p>WARNING: Client will be uninformed about disconnect. Use with caution.
*
* @param id of the client to disconnect
*/
public void disconnect(SessionId id) {
SessionHandle handle = sessions.get(id);
if (handle == null) {
logger.warn("Requested to disconnect client with id {}. Failed as client is not found", id.value());
logger.warn(
"Requested to disconnect client with id {}. Failed as client is not found",
id.value());
return;
}
logger.debug("Disconnecting session {}", id.value());
@@ -29,7 +29,10 @@ public class SessionWriter implements Runnable {
packet = new RawPacket(response.requestId(), response.payload());
transport.write(packet);
} catch (IOException e) {
logger.error("Unexpected exception while writing to transport. RawPacket: {}", packet, e);
logger.error(
"Unexpected exception while writing to transport. RawPacket: {}",
packet,
e);
} catch (InterruptedException e) {
logger.warn("Thread got interrupted", e);
break;