client_connection.cpp

/*
 * Implementation of the ClientConnection class.
 * This class is responsible for handling all interactions with a single
 * connected client, including reading requests, processing them, managing
 * client-specific state, and sending back responses.
 * 
 * CSF Assignment 5
 * 
 * Ifrah Attar - iattar1@jh.edu
 * Morgan Huberty - mhubert3@jh.edu
 * 
 */

 #include <iostream>
 #include <cassert>
 #include <sstream>
 #include <memory>
 #include "csapp.h"
 #include "message.h"
 #include "message_serialization.h"
 #include "server.h"
 #include "exceptions.h"
 #include "client_connection.h"
 #include "guard.h"
 
 /**
  * @brief Constructs a ClientConnection object.
  * @param server A pointer to main Server object.
  * @param client_fd The file descriptor for the client's socket connection.
  */
 ClientConnection::ClientConnection( Server *server, int client_fd )
   : m_server( server )
   , m_client_fd( client_fd )
   , m_logged_in( false )
   , m_in_transaction( false )
 {
   // Initialize a RIO buffer for I/O with the client.
   rio_readinitb( &m_fdbuf, m_client_fd );
 }
 
 /**
  * @brief Destroys the ClientConnection object.
  * Ensures the client's socket is closed to prevent resource leaks.
  */
 ClientConnection::~ClientConnection()
 {
   // The client_fd is checked to prevent closing an invalid descriptor.
   if (m_client_fd >= 0) {
     Close(m_client_fd);
   }
 }
 
 /**
  * @brief The main loop for interacting with a client.
  * This function reads and processes messages from the client until
  * connection is closed or fatal error occurs.
  */
 void ClientConnection::chat_with_client()
 {
   try {
     char buf[Message::MAX_ENCODED_LEN];
     // Loop as long as we can read a line from client.
     while (Rio_readlineb(&m_fdbuf, buf, sizeof(buf)) > 0) {
       Message request;
       try {
         // The inner try block handles recoverable protocol/operation errors.
         MessageSerialization::decode(buf, request);
 
         // Enforce protocol: a client must log in before any other command.
         if (!m_logged_in && request.get_message_type() != MessageType::LOGIN && request.get_message_type() != MessageType::BYE) {
           throw InvalidMessage("Client not logged in");
         }
 
         // Dispatch to appropriate handler based on request type.
         switch (request.get_message_type()) {
           case MessageType::LOGIN: handle_login(request); break;
           case MessageType::CREATE: handle_create(request); break;
           case MessageType::PUSH: handle_push(request); break;
           case MessageType::POP: handle_pop(); break;
           case MessageType::TOP: handle_top(); break;
           case MessageType::SET: handle_set(request); break;
           case MessageType::GET: handle_get(request); break;
           case MessageType::ADD: handle_add(); break;
           case MessageType::SUB: handle_sub(); break;
           case MessageType::MUL: handle_mul(); break;
           case MessageType::DIV: handle_div(); break;
           case MessageType::BEGIN: handle_begin(); break;
           case MessageType::COMMIT: handle_commit(); break;
           case MessageType::BYE:
             send_response(MessageType::OK);
             return; // Exit the loop and close the connection.
           default:
             throw InvalidMessage("Unsupported request type");
         }
       } catch (const InvalidMessage& e) {
         // Malformed or out-of-sequence message is a fatal error.
         send_response(MessageType::ERROR, e.what());
         return; // Terminate connection.
       } catch (const OperationException& e) {
         // A requested operation failed.
         // If this happened during a transaction, we roll it back.
         if (m_in_transaction) rollback_and_release_locks();
         send_response(MessageType::FAILED, e.what());
       } catch (const FailedTransaction& e) {
         // A transaction failed due to a lock issue or nested BEGIN.
         // Rollback is required here.
         rollback_and_release_locks();
         send_response(MessageType::FAILED, e.what());
       }
     }
   } catch (const CommException& e) {
     // The outer try block catches unrecoverable I/O errors.
     m_server->log_error("Communication error with client: " + std::string(e.what()));
   }
   
   // Final check: if the client disconnected while a transaction was active,
   // we must roll back their changes to ensure data interity.
   if (m_in_transaction) {
     rollback_and_release_locks();
   }
 }
 
 void ClientConnection::handle_login(const Message& msg) {
   if (m_logged_in) {
     throw InvalidMessage("Client already logged in");
   }
   m_logged_in = true;
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_create(const Message& msg) {
   // Delegate to the server's thread-safe create_table method.
   m_server->create_table(msg.get_table());
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_push(const Message& msg) {
   // Use the ValueStack class to manage operand stack.
   m_value_stack.push(msg.get_value());
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_pop() {
   m_value_stack.pop(); // Throws OperationException if stack is empty.
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_top() {
   std::string value = m_value_stack.get_top(); // Throws if empty.
   // TOP is the only command that gets a DATA response on success.
   send_response(MessageType::DATA, value);
 }
 
 void ClientConnection::handle_get(const Message& msg) {
   Table* table = m_server->find_table(msg.get_table());
   
   if (m_in_transaction) {
     // In a transaction, use non-blocking lock to prevent deadlocks.
     if (m_locked_tables.find(table) == m_locked_tables.end()) {
       if (!table->trylock()) {
         throw FailedTransaction("Could not acquire lock for GET");
       }
       m_locked_tables.insert(table);
     }
     m_value_stack.push(table->get(msg.get_key()));
   } else {
     // In autocommit mode, use a blocking lock but ensure it is always released.
     table->lock();
     try {
       m_value_stack.push(table->get(msg.get_key()));
       table->unlock();
     } catch (...) {
       // If table->get() throws an exception, make sure to unlock before re-throwing.
       table->unlock();
       throw;
     }
   }
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_set(const Message& msg) {
   std::string value = m_value_stack.get_top();
   m_value_stack.pop();
   
   Table* table = m_server->find_table(msg.get_table());
 
   if (m_in_transaction) {
     // In a transaction, use non-blocking lock.
     if (m_locked_tables.find(table) == m_locked_tables.end()) {
       if (!table->trylock()) {
         throw FailedTransaction("Could not acquire lock for SET");
       }
       m_locked_tables.insert(table);
     }
     table->set(msg.get_key(), value);
   } else {
     // In autocommit mode, lock, set, and immediately commit the change.
     // Use a try/catch block to ensure unlock happens even if there's an error.
     table->lock();
     try {
       table->set(msg.get_key(), value);
       table->commit_changes();
       table->unlock();
     } catch (...) {
       table->unlock();
       throw;
     }
   }
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_begin() {
   if (m_in_transaction) {
     throw FailedTransaction("Nested transactions are not allowed");
   }
   m_in_transaction = true;
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_commit() {
   if (!m_in_transaction) {
     throw OperationException("Not in a transaction");
   }
   for (Table* table : m_locked_tables) {
     table->commit_changes();
     table->unlock();
   }
   m_locked_tables.clear();
   m_in_transaction = false;
   send_response(MessageType::OK);
 }
 
 /**
  * @brief Rolls back changes and releases all locks held by a transaction.
  * This is the primary cleanup function for any failed or aborted transaction.
  */
 void ClientConnection::rollback_and_release_locks() {
   for (Table* table : m_locked_tables) {
     table->rollback_changes();
     table->unlock();
   }
   m_locked_tables.clear();
   m_in_transaction = false;
 }
 
 namespace {
 /**
  * @brief Helper to convert a string to a long, throwing OperationException on failure.
  */
 long string_to_long(const std::string& s) {
   try {
     size_t pos;
     long val = std::stol(s, &pos);
     if (pos != s.length()) {
       throw std::invalid_argument("Invalid integer format");
     }
     return val;
   } catch (...) {
     throw OperationException("Value is not an integer");
   }
 }
 }
 
 void ClientConnection::handle_add() {
   long val2 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   long val1 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   m_value_stack.push(std::to_string(val1 + val2));
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_sub() {
   long val2 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   long val1 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   m_value_stack.push(std::to_string(val1 - val2));
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_mul() {
   long val2 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   long val1 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   m_value_stack.push(std::to_string(val1 * val2));
   send_response(MessageType::OK);
 }
 
 void ClientConnection::handle_div() {
   long val2 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   long val1 = string_to_long(m_value_stack.get_top()); m_value_stack.pop();
   if (val2 == 0) throw OperationException("Division by zero");
   m_value_stack.push(std::to_string(val1 / val2));
   send_response(MessageType::OK);
 }
 
 /**
  * @brief Encapsulates the logic for sending a simple response to the client.
  * @param type The MessageType of the response (e.g., OK, FAILED, ERROR).
  * @param text The optional text payload for FAILED, ERROR, or DATA messages.
  */
 void ClientConnection::send_response(MessageType type, const std::string& text) {
   Message response;
   response.set_message_type(type);
   if (!text.empty()) {
     response.push_arg(text);
   }
   std::string encoded;
   MessageSerialization::encode(response, encoded);
   Rio_writen(m_client_fd, encoded.c_str(), encoded.length());
 }