server_kqueue.cpp

#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/event.h> 
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>        
#include <cstring>
#include <ctime>
#include <map>
#include <thread>
#include <atomic>

using namespace std;

struct Client {
    int socket;
    string name;
};

map<int, Client> clients;  // socket fd -> Client info
int kq;                    // kqueue file descriptor (the event monitor)
int serverSocket;
atomic<bool> running(true);

void printSystem(const string& msg) {
    cout << "\033[33m[System] " << msg << "\033[0m" << endl;
}

void printMessage(const string& sender, const string& msg) {
    time_t now = time(0);
    char timestamp[9];
    strftime(timestamp, sizeof(timestamp), "%H:%M:%S", localtime(&now));
    cout << "\033[90m[" << timestamp << "]\033[0m ";
    cout << "\033[35m" << sender << ":\033[0m ";
    cout << msg << endl;
}

void setNonBlocking(int sock) {
    int flags = fcntl(sock, F_GETFL, 0);        // Get current flags
    fcntl(sock, F_SETFL, flags | O_NONBLOCK);   // Add non-blocking flag
}

void watchSocket(int sock) {
    struct kevent change;
    // EV_SET(event, socket, filter, action, flags, data, userdata)
    // EVFILT_READ = notify when data is available to read
    // EV_ADD = add this socket to watch list
    // EV_ENABLE = enable notifications
    EV_SET(&change, sock, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, NULL);
    kevent(kq, &change, 1, NULL, 0, NULL);  // Apply the change
}

// Stop watching this socket
void unwatchSocket(int sock) {
    struct kevent change;
    EV_SET(&change, sock, EVFILT_READ, EV_DELETE, 0, 0, NULL);
    kevent(kq, &change, 1, NULL, 0, NULL);
}

//messaging
void sendToClient(int sock, const string& msg) {
    send(sock, msg.c_str(), msg.length(), 0);
}

void broadcastMessage(const string& msg, int excludeSocket = -1) {
    for (auto& [sock, client] : clients) {
        if (sock != excludeSocket && !client.name.empty()) {
            sendToClient(sock, msg);
        }
    }
}

string getClientList() {
    string list;
    for (auto& [sock, client] : clients) {
        if (!client.name.empty()) {
            if (!list.empty()) list += ", ";
            list += client.name;
        }
    }
    return list;
}

int findClientByName(const string& name) {
    for (auto& [sock, client] : clients) {
        if (client.name == name) return sock;
    }
    return -1;
}

void acceptNewClient() {
    sockaddr_in clientAddr;
    socklen_t clientLen = sizeof(clientAddr);

    // Accept the connection (non-blocking, returns immediately)
    int clientSock = accept(serverSocket, (sockaddr*)&clientAddr, &clientLen);
    if (clientSock == -1) return;

    // Make client socket non-blocking too
    setNonBlocking(clientSock);

    // Tell kqueue to watch this new client
    watchSocket(clientSock);

    // Add to our clients map (name will be set when they send it)
    clients[clientSock] = {clientSock, ""};

    char ip[INET_ADDRSTRLEN];
    inet_ntop(AF_INET, &clientAddr.sin_addr, ip, INET_ADDRSTRLEN);
    printSystem(string("New connection from ") + ip);
}

void removeClient(int sock) {
    if (clients.count(sock)) {
        string name = clients[sock].name;
        if (!name.empty()) {
            printSystem("'" + name + "' disconnected.");
            broadcastMessage("[" + name + " left the chat]", sock);
        }
        clients.erase(sock);
    }
    unwatchSocket(sock);
    close(sock);
}


void processMessage(int sock, const string& message) {
    Client& client = clients[sock];

    // First message from client is their username
    if (client.name.empty()) {
        client.name = message;
        printSystem("'" + client.name + "' joined the chat!");
        broadcastMessage("[" + client.name + " joined the chat]", sock);
        return;
    }

    // Handle special message types
    if (message.rfind("FILE:", 0) == 0) {
        // File transfer
        size_t pos1 = message.find(':', 5);
        size_t pos2 = message.find(':', pos1 + 1);
        size_t pos3 = message.find(':', pos2 + 1);

        if (pos1 != string::npos && pos2 != string::npos && pos3 != string::npos) {
            string recipient = message.substr(5, pos1 - 5);
            string filename = message.substr(pos1 + 1, pos2 - pos1 - 1);
            string sizeStr = message.substr(pos2 + 1, pos3 - pos2 - 1);
            string fileData = message.substr(pos3 + 1);

            printSystem(client.name + " sending file '" + filename + "' to " + recipient);

            string fullMessage = "FILE:" + client.name + ":" + filename + ":" + sizeStr + ":" + fileData;

            if (recipient == "all") {
                broadcastMessage(fullMessage, sock);
            } else {
                int target = findClientByName(recipient);
                if (target != -1) {
                    sendToClient(target, fullMessage);
                } else {
                    sendToClient(sock, "[System] User '" + recipient + "' not found.");
                }
            }
        }
    }
    else if (message == "USERS") {
        sendToClient(sock, "[Online users: " + getClientList() + "]");
    }
    else if (message.rfind("MSG:", 0) == 0) {
        // Private message
        size_t pos1 = message.find(':', 4);
        if (pos1 != string::npos) {
            string recipient = message.substr(4, pos1 - 4);
            string privateMsg = message.substr(pos1 + 1);

            int target = findClientByName(recipient);
            if (target != -1) {
                sendToClient(target, "[PM from " + client.name + "]: " + privateMsg);
                printSystem(client.name + " -> " + recipient + ": " + privateMsg);
            } else {
                sendToClient(sock, "[System] User '" + recipient + "' not found.");
            }
        }
    }
    else {
        // Regular message - broadcast to all
        printMessage(client.name, message);
        broadcastMessage(client.name + ": " + message, sock);
    }
}

void handleClientData(int sock) {
    char buf[65536];

    // Non-blocking recv - returns immediately
    ssize_t bytes = recv(sock, buf, sizeof(buf), 0);

    if (bytes <= 0) {
        // 0 = client closed connection, -1 = error
        removeClient(sock);
        return;
    }

    string message(buf, bytes);
    processMessage(sock, message);
}

void serverInputThread() {
    char buf[4096];
    while (running) {
        cin.getline(buf, sizeof(buf));
        if (strlen(buf) == 0) continue;

        string input = buf;

        if (input == "/quit" || input == "/q") {
            printSystem("Shutting down...");
            running = false;
            close(serverSocket);
            exit(0);
        }
        else if (input == "/help" || input == "/h") {
            cout << "\n\033[33m========== SERVER COMMANDS ==========\033[0m\n";
            cout << "  /help             Show this help\n";
            cout << "  /quit             Shutdown server\n";
            cout << "  /list             List connected clients\n";
            cout << "  /stats            Show server stats\n";
            cout << "  <message>         Broadcast to all\n";
            cout << "\033[33m=====================================\033[0m\n\n";
        }
        else if (input == "/list" || input == "/l") {
            cout << "\n\033[33mConnected clients (" << clients.size() << "):\033[0m\n";
            for (auto& [sock, client] : clients) {
                if (!client.name.empty()) {
                    cout << "  - " << client.name << " (fd: " << sock << ")\n";
                }
            }
            cout << "\n";
        }
        else if (input == "/stats") {
            cout << "\n\033[33m========== SERVER STATS ==========\033[0m\n";
            cout << "  Model: Event-driven (kqueue)\n";
            cout << "  Threads: 2 (main + input)\n";
            cout << "  Clients: " << clients.size() << "\n";
            cout << "  Could handle: 10,000+ connections\n";
            cout << "\033[33m==================================\033[0m\n\n";
        }
        else {
            string message = "[Server]: " + input;
            broadcastMessage(message);
            printMessage("You (to all)", input);
        }
    }
}


int main() {
    cout << "\033[2J\033[H";
    cout << "\033[35m";
    cout << "================================\n";
    cout << "   EVENT-DRIVEN SERVER v5.0    \n";
    cout << "     (kqueue - single thread)  \n";
    cout << "================================\n";
    cout << "\033[0m\n";

    // Create server socket
    serverSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (serverSocket == -1) {
        cerr << "Can't create socket!" << endl;
        return 1;
    }

    int opt = 1;
    setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(54000);
    addr.sin_addr.s_addr = INADDR_ANY;

    if (::bind(serverSocket, (sockaddr*)&addr, sizeof(addr)) == -1) {
        cerr << "Bind failed!" << endl;
        return 1;
    }

    if (listen(serverSocket, SOMAXCONN) == -1) {
        cerr << "Listen failed!" << endl;
        return 1;
    }

    // Make server socket non-blocking
    setNonBlocking(serverSocket);

    //kqueue
    kq = kqueue();
    if (kq == -1) {
        cerr << "kqueue creation failed!" << endl;
        return 1;
    }

    // Watch server socket for new connections
    watchSocket(serverSocket);

    printSystem("Server started on port 54000");
    printSystem("Mode: Event-driven (kqueue)");
    printSystem("Type /help for commands.\n");

    // Start input thread (for keyboard)
    thread inputThread(serverInputThread);
    inputThread.detach();

   
    struct kevent events[64];  // Buffer for events

    while (running) {
       
        struct timespec timeout = {0, 100000000};  // 100ms timeout
        int numEvents = kevent(kq, NULL, 0, events, 64, &timeout);

        if (numEvents == -1) {
            if (errno == EINTR) continue;  // Interrupted, retry
            break;
        }

        // Process each event
        for (int i = 0; i < numEvents; i++) {
            int fd = events[i].ident;  // Which socket triggered?

            if (events[i].flags & EV_EOF) {
                // Client disconnected
                removeClient(fd);
            }
            else if (fd == serverSocket) {
                // New connection on server socket
                acceptNewClient();
            }
            else {
                // Data from existing client
                handleClientData(fd);
            }
        }
    }

    close(kq);
    close(serverSocket);
    return 0;
}