diff --git a/.gitignore b/.gitignore index c29286e..56624a0 100644 --- a/.gitignore +++ b/.gitignore @@ -121,4 +121,5 @@ $RECYCLE.BIN/ .idea ## bin -bin/ \ No newline at end of file +bin/ +gradle.properties \ No newline at end of file diff --git a/build.gradle b/build.gradle index e126d6d..a7f3f9e 100644 --- a/build.gradle +++ b/build.gradle @@ -55,7 +55,7 @@ checkstyle { spotless { java { - googleJavaFormat('1.25.2').aosp() + googleJavaFormat('1.35.0').aosp() importOrder() removeUnusedImports() trimTrailingWhitespace() diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/ServerApp.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/ServerApp.java index d44df55..126b341 100644 --- a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/ServerApp.java +++ b/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(); - eventBus.subscribe( - DisconnectEvent.class, event -> sessionManager.removeSession(event.sessionId())); - NetworkManager networkManager = new NetworkManager(port, sessionManager, eventBus); + SessionManager sessionManager = new SessionManager(eventBus); + eventBus.subscribe(DisconnectEvent.class, event -> sessionManager.onDisconnect(event)); + NetworkManager networkManager = new NetworkManager(port, sessionManager); UserRegistry userRegistry = new UserRegistry(); eventBus.subscribe( diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/NetworkManager.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/NetworkManager.java index ad2e5b3..d9382a4 100644 --- a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/NetworkManager.java +++ b/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.addSession(session); - session.start(); + sessionManager.create(new TcpTransport(clientSocket)); } } catch (IOException e) { diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/response/PrimitiveResponse.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/response/PrimitiveResponse.java new file mode 100644 index 0000000..38a7422 --- /dev/null +++ b/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) {} diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/Session.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/Session.java index 97e3943..0a9be66 100644 --- a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/Session.java +++ b/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.parser.ProtocolParser; -import ch.unibas.dmi.dbis.cs108.casono.server.network.transport.RawPacket; +import ch.unibas.dmi.dbis.cs108.casono.server.network.response.PrimitiveResponse; 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 org.apache.logging.log4j.Logger; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; /** Represents a client session in the network server. */ -public class Session implements Runnable { - private SessionId id; - private Thread thread; - private TransportLayer transport; - private Logger logger; - private Boolean running; - private EventBus eventBus; +public class Session { + private final SessionId id; + private final TransportLayer transport; + private final BlockingQueue responseQueue; + private static final int RESPOND_QUEUE_SIZE = 32; /** * 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.eventBus = eventBus; - - this.logger = LogManager.getLogger(Session.class.toString() + id.value()); - this.logger.info("Created new session"); + this.responseQueue = new ArrayBlockingQueue<>(RESPOND_QUEUE_SIZE); } /** @@ -47,39 +36,21 @@ public class Session implements Runnable { return this.id; } - /** Starts the session thread. */ - public void start() { - thread.start(); + /** + * Returns the TransportLayer of this session + * + * @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 { - transport.close(); - 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; - } - } + public BlockingQueue getResponseQueue() { + return responseQueue; } } diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionHandle.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionHandle.java new file mode 100644 index 0000000..a603a8a --- /dev/null +++ b/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) {} diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionManager.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionManager.java index 6702543..2ff6e38 100644 --- a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionManager.java +++ b/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 sessions; + private Map 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 + *

Will create both worker threads and start them. + * + * @param transport to create session from + * @return newly created session */ - public void addSession(Session session) { - sessions.put(session.getId(), session); + public Session create(TransportLayer transport) { + 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 - * @return the removed session, or null if not found + *

WARNING: Client will be uninformed about disconnect. Use with caution. + * + * @param id of the client to disconnect */ - public Session removeSession(SessionId id) { - return sessions.remove(id); + public void disconnect(SessionId 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 - * @return the removed session, or null if not found + * @param id of the session that disconnected */ - public Session removeSession(Session session) { - return sessions.remove(session.getId()); + public void onDisconnect(DisconnectEvent event) { + 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(); } } diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionReader.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionReader.java new file mode 100644 index 0000000..7c2a6c7 --- /dev/null +++ b/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); + } + } + } +} diff --git a/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionWriter.java b/src/main/java/ch/unibas/dmi/dbis/cs108/casono/server/network/sessions/SessionWriter.java new file mode 100644 index 0000000..e5c3e78 --- /dev/null +++ b/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 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; + } + } + } +}