Merge remote-tracking branch 'origin/main'

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/ServerApp.java

@@ -26,10 +26,9 @@ public class ServerApp {
         logger.info("Starting server at port {}", port);
 
         EventBus eventBus = new EventBus();
-        SessionManager sessionManager = new SessionManager();
+        SessionManager sessionManager = new SessionManager(eventBus);
-        eventBus.subscribe(
+        eventBus.subscribe(DisconnectEvent.class, event -> sessionManager.onDisconnect(event));
-                DisconnectEvent.class, event -> sessionManager.removeSession(event.sessionId()));
+        NetworkManager networkManager = new NetworkManager(port, sessionManager);
-        NetworkManager networkManager = new NetworkManager(port, sessionManager, eventBus);
 
         UserRegistry userRegistry = new UserRegistry();
         eventBus.subscribe(

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/NetworkManager.java

@@ -1,8 +1,5 @@
 package ch.unibas.dmi.dbis.cs108.casono.server.network;
 
-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.sessions.Session;
 import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.SessionManager;
 import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.TcpTransport;
 import java.io.IOException;
@@ -18,23 +15,19 @@ public class NetworkManager implements Runnable {
     private Thread thread;
     private Boolean running;
     private SessionManager sessionManager;
-    private EventBus eventBus;
 
     /**
      * Creates a new NetworkManager with the given port, session manager, and event bus.
      *
      * @param port the port to listen on
      * @param sessionManager the session manager to use
-     * @param eventBus the event bus for events
      */
-    public NetworkManager(Integer port, SessionManager sessionManager, EventBus eventBus) {
+    public NetworkManager(Integer port, SessionManager sessionManager) {
         this.port = port;
         this.logger = LogManager.getLogger(NetworkManager.class);
         this.thread = new Thread(this, "networkManager");
         this.running = true;
         this.sessionManager = sessionManager;
-        this.eventBus = eventBus;
-        this.eventBus.subscribe(DisconnectEvent.class, event -> clientDisconnected(event));
     }
 
     /** Starts the internal thread to accept new connections. */
@@ -43,15 +36,6 @@ public class NetworkManager implements Runnable {
         thread.start();
     }
 
-    /**
-     * Handles client disconnection events.
-     *
-     * @param event the disconnect event
-     */
-    public void clientDisconnected(DisconnectEvent event) {
-        logger.info("Session {} disconnected", event.sessionId().value());
-    }
-
     /** Runs the network manager loop, accepting connections. */
     @Override
     public void run() {
@@ -61,9 +45,7 @@ public class NetworkManager implements Runnable {
 
                 logger.debug("Accepted connection from {}", clientSocket.getRemoteSocketAddress());
 
-                Session session = new Session(new TcpTransport(clientSocket), eventBus);
+                sessionManager.create(new TcpTransport(clientSocket));
-                sessionManager.addSession(session);
-                session.start();
             }
 
         } catch (IOException e) {

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/response/PrimitiveResponse.java

@@ -0,0 +1,5 @@
+package ch.unibas.dmi.dbis.cs108.casono.server.network.response;
+
+import ch.unibas.dmi.dbis.cs108.casono.server.network.sessions.SessionId;
+
+public record PrimitiveResponse(SessionId sessionId, int requestId, String payload) {}

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/Session.java

@@ -1,24 +1,18 @@
 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.parser.PrimitiveRequest;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.response.PrimitiveResponse;
-import ch.unibas.dmi.dbis.cs108.casono.server.network.parser.ProtocolParser;
-import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.RawPacket;
 import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.TransportLayer;
-import java.io.EOFException;
 import java.io.IOException;
-import org.apache.logging.log4j.LogManager;
+import java.util.concurrent.ArrayBlockingQueue;
-import org.apache.logging.log4j.Logger;
+import java.util.concurrent.BlockingQueue;
 
 /** Represents a client session in the network server. */
-public class Session implements Runnable {
+public class Session {
-    private SessionId id;
+    private final SessionId id;
-    private Thread thread;
+    private final TransportLayer transport;
-    private TransportLayer transport;
+    private final BlockingQueue<PrimitiveResponse> responseQueue;
-    private Logger logger;
+    private static final int RESPOND_QUEUE_SIZE = 32;
-    private Boolean running;
-    private EventBus eventBus;
 
     /**
      * Creates a new Session with the given transport and event bus.
@@ -27,15 +21,10 @@ public class Session implements Runnable {
      * @param eventBus the event bus for publishing events
      * @throws IOException if an I/O error occurs during initialization
      */
-    public Session(TransportLayer transport, EventBus eventBus) throws IOException {
+    public Session(TransportLayer transport, EventBus eventBus) {
         this.id = new SessionId();
-        this.thread = new Thread(this, "session-" + this.id.value());
         this.transport = transport;
-        this.running = true;
+        this.responseQueue = new ArrayBlockingQueue<>(RESPOND_QUEUE_SIZE);
-        this.eventBus = eventBus;
-
-        this.logger = LogManager.getLogger(Session.class.toString() + id.value());
-        this.logger.info("Created new session");
     }
 
     /**
@@ -47,39 +36,21 @@ public class Session implements Runnable {
         return this.id;
     }
 
-    /** Starts the session thread. */
+    /**
-    public void start() {
+     * Returns the TransportLayer of this session
-        thread.start();
+     *
+     * @return the transport layer of the session
+     */
+    public TransportLayer getTransport() {
+        return transport;
     }
 
     /**
-     * Closes the session and its transport.
+     * Returns the BlockingQueue of this session
      *
-     * @throws IOException if an I/O error occurs
+     * @return the queue containing outgoing responses
      */
-    public void close() throws IOException {
+    public BlockingQueue<PrimitiveResponse> getResponseQueue() {
-        transport.close();
+        return responseQueue;
-        this.running = false;
-    }
-
-    /** Runs the session loop, reading from the transport. */
-    @Override
-    public void run() {
-        while (running) {
-            try {
-                RawPacket rawPacket = transport.read();
-                logger.debug("Recieved: {}", rawPacket);
-
-                PrimitiveRequest primitiveRequest = ProtocolParser.parse(rawPacket);
-                logger.debug("Parsed request to {}", primitiveRequest);
-            } catch (EOFException e) {
-                logger.info("Client disconnected");
-                eventBus.publish(new DisconnectEvent(id));
-                break;
-            } catch (IOException e) {
-                e.printStackTrace();
-                break;
-            }
-        }
     }
 }

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionHandle.java

@@ -0,0 +1,4 @@
+package ch.unibas.dmi.dbis.cs108.casono.server.network.sessions;
+
+/** The SessionHandle stores the session and the two worker threads associated with the session */
+record SessionHandle(Session session, Thread reader, Thread writer) {}

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionManager.java

@@ -1,44 +1,88 @@
 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;
 
 /** Manages active sessions in the server. */
 public class SessionManager {
-    private Map<SessionId, Session> sessions;
+    private Map<SessionId, SessionHandle> sessions;
+    private final EventBus eventBus;
+    private final Logger logger;
 
     /** Constructs a new SessionManager. */
-    public SessionManager() {
+    public SessionManager(EventBus eventBus) {
         this.sessions = new ConcurrentHashMap<>();
+        this.eventBus = eventBus;
+        this.logger = LogManager.getLogger(SessionManager.class);
     }
 
     /**
-     * Adds a session to the manager.
+     * Create new Session from provided transport.
      *
-     * @param session the session to add
+     * <p>Will create both worker threads and start them.
+     *
+     * @param transport to create session from
+     * @return newly created session
      */
-    public void addSession(Session session) {
+    public Session create(TransportLayer transport) {
-        sessions.put(session.getId(), session);
+        Session session = new Session(transport, eventBus);
+        SessionReader reader = new SessionReader(session, eventBus);
+        SessionWriter writer = new SessionWriter(session);
+
+        Thread readerThread = Thread.ofVirtual()
+            .name("session-" + session.getId().value() + "-reader")
+            .unstarted(reader);
+        Thread writerThread = Thread.ofVirtual()
+            .name("session-" + session.getId().value() + "-writer")
+            .unstarted(writer);
+
+        sessions.put(session.getId(), new SessionHandle(session, readerThread, writerThread));
+        readerThread.start();
+        writerThread.start();
+        return session;
     }
 
     /**
-     * Removes a session by its ID.
+     * Disconnect specified client
      *
-     * @param id the ID of the session to remove
+     * <p>WARNING: Client will be uninformed about disconnect. Use with caution.
-     * @return the removed session, or null if not found
+     *
+     * @param id of the client to disconnect
      */
-    public Session removeSession(SessionId id) {
+    public void disconnect(SessionId id) {
-        return sessions.remove(id);
+        SessionHandle handle = sessions.remove(id);
+        if (handle == null) {
+            logger.warn(
+                    "Requested to disconnect client with id {}. Failed as client is not found",
+                    id.value());
+            return;
+        }
+        logger.debug("Disconnecting session {}", id.value());
+
+        handle.reader().interrupt();
+        handle.writer().interrupt();
+        try {
+            handle.session().getTransport().close();
+        } catch (IOException e) {
+            logger.error("Unexpected exception while closing transport", e);
+        }
     }
 
     /**
-     * Removes the specified session.
+     * Handler for the DisconnectEvent
      *
-     * @param session the session to remove
+     * @param id of the session that disconnected
-     * @return the removed session, or null if not found
      */
-    public Session removeSession(Session session) {
+    public void onDisconnect(DisconnectEvent event) {
-        return sessions.remove(session.getId());
+        logger.debug("Recieved DisconnectEvent event for session {}", event.sessionId().value());
+
+        disconnect(event.sessionId());
     }
 
     /**
@@ -48,6 +92,12 @@ public class SessionManager {
      * @return the session with the specified ID, or null if not found
      */
     public Session getSessionById(SessionId id) {
-        return sessions.get(id);
+        SessionHandle handle = sessions.get(id);
+
+        if (handle == null) {
+            return null;
+        }
+
+        return handle.session();
     }
 }

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionReader.java

@@ -0,0 +1,54 @@
+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.parser.PrimitiveRequest;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.parser.ProtocolParser;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.parser.ProtocolParserException;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.RawPacket;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.TransportLayer;
+import ch.unibas.dmi.dbis.cs108.casono.server.tokenizer.TokenizerException;
+import java.io.EOFException;
+import java.io.IOException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class SessionReader implements Runnable {
+    private final Session session;
+    private final TransportLayer transport;
+    private final EventBus eventBus;
+    private final Logger logger;
+
+    public SessionReader(Session session, EventBus eventBus) {
+        this.session = session;
+        this.transport = session.getTransport();
+        this.eventBus = eventBus;
+        this.logger =
+                LogManager.getLogger(
+                        SessionReader.class.toString() + "-" + session.getId().value());
+    }
+
+    @Override
+    public void run() {
+        while (!Thread.currentThread().isInterrupted()) {
+            RawPacket rawPacket = null;
+            try {
+                rawPacket = transport.read();
+                logger.debug("Recieved: {}", rawPacket);
+
+                PrimitiveRequest primitiveRequest = ProtocolParser.parse(rawPacket);
+                logger.debug("Parsed request to {}", primitiveRequest);
+            } catch (EOFException e) {
+                logger.info("Client disconnected");
+                eventBus.publish(new DisconnectEvent(session.getId()));
+                break;
+            } catch (TokenizerException | ProtocolParserException e) {
+                logger.error("Error occured while parsing request. RawPacket: {}", rawPacket, e);
+
+                // TODO: Send error response to client
+            } catch (IOException e) {
+                logger.error("Unexpected exception while reading from transport", e);
+            }
+        }
+    }
+}

src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionWriter.java

@@ -0,0 +1,42 @@
+package ch.unibas.dmi.dbis.cs108.casono.server.network.sessions;
+
+import ch.unibas.dmi.dbis.cs108.casono.server.network.response.PrimitiveResponse;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.RawPacket;
+import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.TransportLayer;
+import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class SessionWriter implements Runnable {
+    private final TransportLayer transport;
+    private final BlockingQueue<PrimitiveResponse> queue;
+    private final Logger logger;
+
+    public SessionWriter(Session session) {
+        this.transport = session.getTransport();
+        this.queue = session.getResponseQueue();
+        this.logger =
+                LogManager.getLogger(
+                        SessionReader.class.toString() + "-" + session.getId().value());
+    }
+
+    public void run() {
+        while (!Thread.currentThread().isInterrupted()) {
+            RawPacket packet = null;
+            try {
+                PrimitiveResponse response = queue.take();
+                packet = new RawPacket(response.requestId(), response.payload());
+                transport.write(packet);
+            } catch (IOException e) {
+                logger.error(
+                        "Unexpected exception while writing to transport. RawPacket: {}",
+                        packet,
+                        e);
+            } catch (InterruptedException e) {
+                logger.warn("Thread got interrupted", e);
+                break;
+            }
+        }
+    }
+}