Developing a concurrent chat server involves managing multiple simultaneous TCP connections while ensuring thread-safe message distribution. This implementation leverages the Winsock2 API alongside C++11 concurrency primitives to build a multi-user messaging platform supporting both direct messaging and broadcast capabilities.

Protocol Design

The application layer protocol uses a simple text-based command structure with comma-separated fields:

COMMAND,sender,target,content

Supported command types:

• REG: Register a username with the server
• MSG: Send a direct message to a specific user
• BRD: Broadcast a message to all connected users

Server Architecture

The server utilizes a thread-per-connection model paired with a centralized message dispatch queue. Incoming network data gets parsed in to structured packets and processed through a producer-consumer pattern, ensuring serialized access to shared user registries.

Data Structures and Globals

#include <winsock2.h>
#include <ws2tcpip.h>
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <unordered_map>
#include <vector>
#include <memory>
#include <sstream>
#include <algorithm>

#pragma comment(lib, "ws2_32.lib")

struct Packet {
    enum Category { REGISTRATION, DIRECT, BROADCAST };
    Category type;
    std::string origin;
    std::string recipient;
    std::string body;
};

class ClientHandler;

std::mutex registryLock;
std::unordered_map<std::string, std::shared_ptr<ClientHandler>> activeUsers;
std::vector<std::shared_ptr<ClientHandler>> connectionPool;

std::queue<std::shared_ptr<Packet>> dispatchQueue;
std::mutex queueLock;
std::condition_variable queueSignal;

std::vector<std::string> tokenize(const std::string& str, char delim) {
    std::vector<std::string> tokens;
    std::stringstream ss(str);
    std::string token;
    while (std::getline(ss, token, delim)) {
        tokens.push_back(token);
    }
    return tokens;
}

void processRegistration(std::shared_ptr<Packet> pkt);
void processDirect(std::shared_ptr<Packet> pkt);
void processBroadcast(std::shared_ptr<Packet> pkt);

Connection Handler Class

class ClientHandler : public std::enable_shared_from_this<ClientHandler> {
    SOCKET sock;
    std::string handle;
    std::thread worker;
    char buffer[2048];
    std::mutex transmitLock;

public:
    explicit ClientHandler(SOCKET socket) : sock(socket) {}
    
    ~ClientHandler() {
        if (worker.joinable()) worker.detach();
    }

    void launch() {
        worker = std::thread(&ClientHandler::receiveLoop, shared_from_this());
    }

    void transmit(const std::string& data) {
        std::lock_guard<std::mutex> guard(transmitLock);
        send(sock, data.c_str(), static_cast<int>(data.length()), 0);
    }

    void setHandle(const std::string& name) { handle = name; }
    std::string getHandle() const { return handle; }

private:
    void receiveLoop() {
        while (true) {
            int received = recv(sock, buffer, sizeof(buffer) - 1, 0);
            if (received <= 0) break;
            
            buffer[received] = '\0';
            parseCommand(buffer);
        }
    }

    void parseCommand(const char* raw) {
        auto fields = tokenize(raw, ',');
        if (fields.size() < 4) {
            transmit("ERR:Invalid format. Use: CMD,sender,target,message");
            return;
        }

        auto pkt = std::make_shared<Packet>();
        pkt->origin = fields[1];

        if (fields[0] == "REG") {
            pkt->type = Packet::REGISTRATION;
            handle = fields[1];
        } else if (fields[0] == "MSG") {
            pkt->type = Packet::DIRECT;
            pkt->recipient = fields[2];
            pkt->body = fields[3];
        } else if (fields[0] == "BRD") {
            pkt->type = Packet::BROADCAST;
            pkt->body = fields[3];
        } else {
            transmit("ERR:Unknown command");
            return;
        }

        {
            std::lock_guard<std::mutex> guard(queueLock);
            dispatchQueue.push(pkt);
        }
        queueSignal.notify_one();
    }
};

Main Server Implementation

int main() {
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {
        std::cerr << "WSAStartup failed\n";
        return 1;
    }

    SOCKET listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (listener == INVALID_SOCKET) {
        WSACleanup();
        return 1;
    }

    sockaddr_in bindAddr{};
    bindAddr.sin_family = AF_INET;
    bindAddr.sin_addr.s_addr = INADDR_ANY;
    bindAddr.sin_port = htons(12345);

    if (bind(listener, (sockaddr*)&bindAddr, sizeof(bindAddr)) == SOCKET_ERROR) {
        closesocket(listener);
        WSACleanup();
        return 1;
    }

    listen(listener, SOMAXCONN);
    std::cout << "Server listening on port 12345...\n";

    // Acceptor thread
    std::thread([&] {
        while (true) {
            sockaddr_in clientAddr;
            int addrLen = sizeof(clientAddr);
            SOCKET clientSock = accept(listener, (sockaddr*)&clientAddr, &addrLen);
            
            if (clientSock == INVALID_SOCKET) continue;

            char ipStr[INET_ADDRSTRLEN];
            inet_ntop(AF_INET, &clientAddr.sin_addr, ipStr, INET_ADDRSTRLEN);
            std::cout << "Connection from " << ipStr << ":" << ntohs(clientAddr.sin_port) << "\n";

            auto handler = std::make_shared<ClientHandler>(clientSock);
            {
                std::lock_guard<std::mutex> guard(registryLock);
                connectionPool.push_back(handler);
            }
            handler->launch();
        }
    }).detach();

    // Dispatcher thread
    std::thread([&] {
        while (true) {
            std::unique_lock<std::mutex> guard(queueLock);
            queueSignal.wait(guard, [] { return !dispatchQueue.empty(); });

            while (!dispatchQueue.empty()) {
                auto pkt = dispatchQueue.front();
                dispatchQueue.pop();
                guard.unlock();

                switch (pkt->type) {
                    case Packet::REGISTRATION: processRegistration(pkt); break;
                    case Packet::DIRECT: processDirect(pkt); break;
                    case Packet::BROADCAST: processBroadcast(pkt); break;
                }
                
                guard.lock();
            }
        }
    }).detach();

    std::string cmd;
    while (std::cin >> cmd) {
        if (cmd == "quit") break;
    }

    closesocket(listener);
    WSACleanup();
    return 0;
}

void processRegistration(std::shared_ptr<Packet> pkt) {
    std::lock_guard<std::mutex> guard(registryLock);
    
    auto it = std::find_if(connectionPool.begin(), connectionPool.end(),
        [&](const auto& h) { return h->getHandle() == pkt->origin; });
        
    if (it == connectionPool.end()) return;

    auto result = activeUsers.insert({pkt->origin, *it});
    (*it)->transmit(result.second ? "SYS:Registration successful" : "ERR:Username taken");
}

void processDirect(std::shared_ptr<Packet> pkt) {
    std::lock_guard<std::mutex> guard(registryLock);
    auto it = activeUsers.find(pkt->recipient);
    if (it != activeUsers.end()) {
        it->second->transmit("[" + pkt->origin + "]: " + pkt->body);
    }
}

void processBroadcast(std::shared_ptr<Packet> pkt) {
    std::lock_guard<std::mutex> guard(registryLock);
    for (auto& [name, handler] : activeUsers) {
        if (name != pkt->origin) {
            handler->transmit("[Broadcast " + pkt->origin + "]: " + pkt->body);
        }
    }
}

Client Implementation

The client maintains a dedicated thread for receiving server transmissions while the main thread handles user input.

#include <winsock2.h>
#include <ws2tcpip.h>
#include <iostream>
#include <thread>
#include <string>

#pragma comment(lib, "ws2_32.lib")

int main() {
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) return 1;

    SOCKET sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (sock == INVALID_SOCKET) return 1;

    sockaddr_in srvAddr{};
    srvAddr.sin_family = AF_INET;
    inet_pton(AF_INET, "127.0.0.1", &srvAddr.sin_addr);
    srvAddr.sin_port = htons(12345);

    if (connect(sock, (sockaddr*)&srvAddr, sizeof(srvAddr)) == SOCKET_ERROR) {
        closesocket(sock);
        WSACleanup();
        return 1;
    }

    std::thread([&] {
        char buffer[1024];
        while (true) {
            int len = recv(sock, buffer, sizeof(buffer) - 1, 0);
            if (len <= 0) break;
            buffer[len] = '\0';
            std::cout << buffer << "\n";
        }
    }).detach();

    std::string input;
    while (std::getline(std::cin, input)) {
        if (input == "exit") break;
        send(sock, input.c_str(), static_cast<int>(input.length()), 0);
    }

    closesocket(sock);
    WSACleanup();
    return 0;
}

Verification Steps

Launch the server, then connect three client instances:

Client A registration:

REG,Alice,,

Response: SYS:Registration successful

Client B registration:

REG,Bob,,

Client C registration:

REG,Charlie,,

Direct message test: From Client A:

MSG,Alice,Bob,Private hello

Client B receives: [Alice]: Private hello

Broadcast test: From Client A:

BRD,Alice,,Hello everyone

Client B receives: [Broadcast Alice]: Hello everyone Client C receives: [Broadcast Alice]: Hello everyone

Critical Implementation Details

This example focuses on concurrency patterns and basic socket I/O. Production implementations must address TCP stream semantics: the protocol lacks message framing mechanisms such as length-prefixing or delimiter handling, which are essential to handle partial reads and packet coalescing (common known as "sticky packet" problems) inherent in stream-based protocols.