Server/SourceFiles/MainServer/Server.cpp - server - Shymer123/Messenger

Server/SourceFiles/MainServer/Server.cpp
321 строка10 KB
Shymer123
compiler warnings removed
30 мар 2025, 08:30
30 мар 2025, 08:30e6e7e55
#include "Server.h"
#include "ServerCache/ServerCache.h"
#include "Logger/Logger.h"
#include "ClientHandler/ClientHandler.h"
#include "DatabaseManager/DatabaseManager.h"
#include <iostream>
#include <fcntl.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>
#include <netinet/tcp.h>
#include <netinet/in.h>
#include <thread>
#include <fstream>

Server::Server(int port) : serverSocket(-1), port(port), maxFd(-1)
{
    FD_ZERO(&masterSet);
    FD_ZERO(&readSet);
}

Server::~Server()
{
    stop();
}

void Server::setupSocket()
{
    Logger::getInstance().logError("Creating server socket...", Logger::logLevel::Info);
    std::cout << "Creating server socket..." << std::endl;

    serverSocket = socket(AF_INET, SOCK_STREAM, 0);

    if(serverSocket < 0)
    {
        Logger::getInstance().logError("General Error in setupSocket: Socket creation failed", Logger::logLevel::Error);
        exit(EXIT_FAILURE);
    }
    Logger::getInstance().logError("Server socket created successfully", Logger::logLevel::Info);
    std::cout << "Server socket created successfully" << std::endl;

    fcntl(serverSocket, F_SETFL, O_NONBLOCK);
    Logger::getInstance().logError("Server socket set to non-blocking", Logger::logLevel::Info);
    std::cout << "Server socket set to non-blocking" << std::endl;

    sockaddr_in serverAddress;
    serverAddress.sin_family = AF_INET;
    serverAddress.sin_port = htons(port);
    serverAddress.sin_addr.s_addr = INADDR_ANY;

    if(bind(serverSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0)
    {
        Logger::getInstance().logError("General Error in setupSocket: Socket bind failed", Logger::logLevel::Error);
        close(serverSocket);
        exit(EXIT_FAILURE);
    }
    Logger::getInstance().logError("Socket bound to port " + std::to_string(port), Logger::logLevel::Info);
    std::cout << "Socket bound to port " + std::to_string(port) << std::endl;

    if(listen(serverSocket, SOMAXCONN) < 0)
    {
        Logger::getInstance().logError("General Error in setupSocket: Socket listen failed", Logger::logLevel::Error);
        close(serverSocket);
        exit(EXIT_FAILURE);
    }
    Logger::getInstance().logError("Server listening on port " + std::to_string(port), Logger::logLevel::Info);
    std::cout << "Server listening on port " + std::to_string(port) << std::endl;

    FD_SET(serverSocket, &masterSet);
    maxFd = serverSocket;
}

void Server::start()
{
    Logger::getInstance().clearFile();

    running = true;
    setupSocket();

    std::ifstream inputFile("config.json");
    json config;
    inputFile >> config;
    inputFile.close();

    const json& serverConfig = config["server"];
    uint16_t numThreads = serverConfig.value("number_threads_process_request", 1);

    for(int i = 0; i < numThreads; ++i)
    {
        workerThreads.emplace_back([this]() { processQueue(); });
    }

    Logger::getInstance().logError("Server started on port " + std::to_string(port), Logger::logLevel::Info);
    std::cout << "Server started on port " + std::to_string(port) << std::endl;

    std::thread([this]()
    {
        std::string command;
        while(running)
        {
            std::getline(std::cin, command);
            if(command == "stop")
            {
                Logger::getInstance().logError("Stopping server on command", Logger::logLevel::Info);
                std::cout << "Stopping server on command" << std::endl;
                running = false;
            }
        }
    }).detach();

    DatabaseManager& dbManager = DatabaseManager::getInstance();
    dbManager.setServerPointer(this);

    while(running)
    {
        readSet = masterSet;

        struct timeval timeout;
        timeout.tv_sec = 0;
        timeout.tv_usec = 100000;

        int activity = select(maxFd + 1, &readSet, nullptr, nullptr, &timeout);

        if(activity < 0)
        {
            Logger::getInstance().logError("General Error in start: Select failed", Logger::logLevel::Error);
            std::cout << "General Error in start: Select failed" << std::endl;
            break;
        }

        if(FD_ISSET(serverSocket, &readSet))
        {
            handleNewConnection();
        }

        handleClientMessages();
    }

    stop();
}

void Server::handleNewConnection()
{
    sockaddr_in clientAddress;
    socklen_t clientAddressSize = sizeof(clientAddress);
    int clientSocket = accept(serverSocket, (struct sockaddr*)&clientAddress, &clientAddressSize);

    if(clientSocket >= 0)
    {
        fcntl(clientSocket, F_SETFL, O_NONBLOCK);
        FD_SET(clientSocket, &masterSet);

        if(clientSocket > maxFd)
        {
            maxFd = clientSocket;
        }

        {
            std::lock_guard<std::mutex> clientsLock(clientsMutex);
            clients.insert(clientSocket);
        }

        {
            std::lock_guard<std::mutex> lock(socketMapMutex);
            socketMutexes.emplace(clientSocket, std::make_unique<std::mutex>());
        }

        Logger::getInstance().logError("New client connected: " +
                                       std::string(inet_ntoa(clientAddress.sin_addr)) +
                                       ":" + std::to_string(ntohs(clientAddress.sin_port)), Logger::logLevel::Info);
        std::cout << "New client connected: " +
                         std::string(inet_ntoa(clientAddress.sin_addr)) +
                         ":" + std::to_string(ntohs(clientAddress.sin_port)) << std::endl;
    }
}

void Server::handleClientMessages()
{
    for(auto it = clients.begin(); it != clients.end() ; )
    {
        int clientSocket = *it;
        if(FD_ISSET(clientSocket, &readSet))
        {
            ClientHandler handler(clientSocket, this);
            std::vector<char> requestData = handler.receiveRequest();
            if(requestData.empty())
            {
                close(clientSocket);
                FD_CLR(clientSocket, &masterSet);

                {
                    std::lock_guard<std::mutex> clientsLock(clientsMutex);
                    it = clients.erase(it);
                }

                ServerCache::getInstance().removeUserInServerCache(clientSocket);

                {
                    std::lock_guard<std::mutex> lock(socketMapMutex);
                    socketMutexes.erase(clientSocket);
                }
                continue;
            }

            enqueueRequest(clientSocket, std::move(requestData));
        }

        ++it;
    }
}

void Server::enqueueRequest(int clientSocket, const std::vector<char> request)
{
    std::lock_guard<std::mutex> lock(queueMutex);
    requestQueue.push({clientSocket, std::move(request)});
    queueCondition.notify_one();
}

void Server::processQueue()
{
    while(running)
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        queueCondition.wait(lock, [this]()
        {
            return !requestQueue.empty() || !running;
        });

        if(!running) return;

        std::pair<int, std::vector<char>> request;
        request = std::move(requestQueue.front());
        requestQueue.pop();

        lock.unlock();

        std::lock_guard<std::mutex> serverLock(serverMutex);
        ClientHandler handler(request.first, this);
        handler.processRequest(request.second);
    }
}

void Server::safeSend(int clientSocket, const char& type, const std::vector<char>& response)
{
    std::lock_guard<std::mutex> mapLock(socketMapMutex);
    std::mutex& socketMutex = *socketMutexes[clientSocket];

    std::lock_guard<std::mutex> socketLock(socketMutex);

    ssize_t bytesTypeSent = send(clientSocket, &type, sizeof(type), 0);
    if(bytesTypeSent <= 0)
    {
        Logger::getInstance().logError("Failed to send type message: " + std::to_string(errno), Logger::logLevel::Error);
        return;
    }

    if(type == 0x01)
    {
        uint32_t responseSize = static_cast<uint32_t>(response.size());
        responseSize = htonl(responseSize);

        ssize_t bytesSent = send(clientSocket, reinterpret_cast<char*>(&responseSize), sizeof(responseSize), 0);
        if (bytesSent <= 0)
        {
            Logger::getInstance().logError("Failed to send response size: " + std::to_string(errno), Logger::logLevel::Error);
            return;
        }
    }

    size_t totalBytesSent = 0;
    const size_t chunkSize = 8192;
    while (totalBytesSent < response.size())
    {
        size_t bytesToSend = std::min(chunkSize, response.size() - totalBytesSent);
        ssize_t bytesSent = send(clientSocket, response.data() + totalBytesSent, bytesToSend, 0);

        if (bytesSent == -1)
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                Logger::getInstance().logError("Socket buffer full, retrying...", Logger::logLevel::Error);
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                continue;
            }
            Logger::getInstance().logError("Failed to send response data: " + std::to_string(errno), Logger::logLevel::Error);
            return;
        }
        if (bytesSent == 0)
        {
            Logger::getInstance().logError("Connection closed by client", Logger::logLevel::Error);
            return;
        }

        totalBytesSent += bytesSent;
        Logger::getInstance().logError("Sent chunk: " + std::to_string(bytesSent) + " bytes, total: " + std::to_string(totalBytesSent), Logger::logLevel::Info);
    }

    Logger::getInstance().logError("Total sent: " + std::to_string(totalBytesSent) + " bytes", Logger::logLevel::Info);
}

void Server::stop()
{
    queueCondition.notify_all();

    for(std::thread& thread : workerThreads)
    {
        if(thread.joinable())
        {
            thread.join();
        }
    }

    for(int client : clients)
    {
        close(client);
        FD_CLR(client, &masterSet);
        ServerCache::getInstance().removeUserInServerCache(client);
    }
    close(serverSocket);
    exit(1);
}