Kerangka server dari bagian 1 sudah mampu menerima koneksi, tetapi masih diam: belum ada yang membaca pesan, dan belum ada tempat menyimpan daftar client. Bagian 2 menghidupkannya dengan model thread per client — setiap koneksi dilayani satu thread penerima yang membaca baris pesan dan menyebarkannya ke client lain lewat registri bersama. Model ini paling mudah dipahami untuk skala kecil dan menjadi dasar untuk memahami model yang lebih canggih seperti event loop Asio pada chapter pengantar nanti.

Ada satu jebakan yang kita hindari sejak awal: bila sebuah client dikirim pesan saat client lain belum masuk daftar, siaran itu lewat begitu saja dan tidak pernah sampai. Karena itu server menerima DUA koneksi lebih dulu, baru menyalakan thread penerima — daftar koneksi sudah lengkap sebelum satu pun pesan dibaca. Sisi client pun menambahkan langkah serupa: keduanya menunggu sampai saling terhubung (rendezvous) sebelum mulai berkirim.

Program Lengkap Bagian Ini #

Berikut chat-server.cpp versi lengkap. Program ini mandiri: server dan dua client berjalan dalam satu proses, output hanya dicetak thread utama setelah semua thread selesai, sehingga hasilnya deterministik dan bisa dibandingkan byte-per-byte:

cpp 
// chat-server.cpp — server chat TCP localhost: satu thread penerima per
// client, broadcast pesan ke client lain, shutdown bersih.
// Skenario dijalankan di satu proses: server + dua client, output hanya
// dicetak thread utama setelah semua thread di-join (deterministik).
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <condition_variable>
#include <cstdint>
#include <cstring>
#include <iostream>
#include <mutex>
#include <optional>
#include <string>
#include <thread>
#include <vector>

namespace {

// Mengirim seluruh byte; send() boleh berhenti di tengah buffer.
bool kirimSemua(int fd, const std::string& pesan) {
    std::size_t terkirim = 0;
    while (terkirim < pesan.size()) {
        const ssize_t n =
            ::send(fd, pesan.data() + terkirim, pesan.size() - terkirim, 0);
        if (n < 0) {
            return false;
        }
        terkirim += static_cast<std::size_t>(n);
    }
    return true;
}

// Membaca satu baris sampai '\n'; false saat koneksi ditutup atau galat.
bool terimaBaris(int fd, std::string& baris) {
    baris.clear();
    char c = 0;
    while (true) {
        const ssize_t n = ::recv(fd, &c, 1, 0);
        if (n <= 0) {
            return false;
        }
        if (c == '\n') {
            return true;
        }
        baris.push_back(c);
    }
}

struct PesanChat {
    std::string pengirim;
    std::string isi;
};

// Mengurai "PENGIRIM|ISI"; nullopt untuk baris perintah '/' atau salah bentuk.
std::optional<PesanChat> parseBaris(const std::string& baris) {
    if (baris.empty() || baris[0] == '/') {
        return std::nullopt;
    }
    const std::size_t posisi = baris.find('|');
    if (posisi == std::string::npos) {
        return std::nullopt;
    }
    PesanChat pesan{baris.substr(0, posisi), baris.substr(posisi + 1)};
    if (pesan.pengirim.empty() || pesan.isi.empty()) {
        return std::nullopt;
    }
    return pesan;
}

// Registri client yang dipakai bersama semua thread penerima.
struct Registri {
    std::mutex kunci;
    std::vector<int> fds;            // koneksi aktif
    std::vector<PesanChat> siaran;   // hasil broadcast

    void tambah(int fd) {
        std::lock_guard<std::mutex> gembok(kunci);
        fds.push_back(fd);
    }

    // Mengirim pesan ke semua client KECUALI pengirimnya.
    void siarkan(const PesanChat& pesan, int fdPengirim) {
        std::lock_guard<std::mutex> gembok(kunci);
        for (int fd : fds) {
            if (fd != fdPengirim) {
                kirimSemua(fd, pesan.pengirim + "|" + pesan.isi + "\n");
            }
        }
        siaran.push_back(pesan);
    }
};

// Satu thread penerima per client: membaca baris dan menyebarkannya.
void layaniClient(int fd, Registri& registri) {
    std::string baris;
    while (terimaBaris(fd, baris)) {
        if (baris == "/quit") {
            break;                    // client pamit pulang
        }
        const std::optional<PesanChat> pesan = parseBaris(baris);
        if (pesan.has_value()) {
            registri.siarkan(*pesan, fd);
        }
    }
    ::close(fd);                      // rapikan koneksi sendiri
}

}  // namespace

int main() {
    // 1) Socket pendengar: socket -> bind -> listen di loopback, port 0.
    const int fdListen = ::socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in alamat{};
    alamat.sin_family = AF_INET;
    alamat.sin_port = htons(0);                       // OS memilih port bebas
    alamat.sin_addr.s_addr = htonl(INADDR_LOOPBACK);  // 127.0.0.1
    if (::bind(fdListen, reinterpret_cast<const sockaddr*>(&alamat),
               sizeof(alamat)) < 0 ||
        ::listen(fdListen, 8) < 0) {
        std::cerr << "bind/listen gagal: " << std::strerror(errno) << "\n";
        return 1;
    }

    // Port yang benar dibaca lewat getsockname, lalu diberi ke client.
    sockaddr_in terikat{};
    socklen_t panjangTerikat = sizeof(terikat);
    ::getsockname(fdListen, reinterpret_cast<sockaddr*>(&terikat),
                  &panjangTerikat);
    const uint16_t port = terikat.sin_port;   // sudah network byte order
    std::cout << "[server] siap di 127.0.0.1, port ephemeral dari getsockname\n";

    Registri registri;

    // Rendezvous: kedua client menunggu sampai keduanya terhubung, baru
    // ada yang mulai mengirim — siaran tidak ada yang terlewat.
    int terhubung = 0;
    std::mutex kunciHubung;
    std::condition_variable sinyalHubung;
    auto tungguBersama = [&] {
        {
            std::lock_guard<std::mutex> gembok(kunciHubung);
            ++terhubung;
        }
        sinyalHubung.notify_all();   // bangunkan pasangan yang mungkin menunggu
        std::unique_lock<std::mutex> gembok(kunciHubung);
        // Thread kedua langsung lolos (predikat sudah true); thread pertama
        // menunggu hingga dibangunkan. Predikat menutup celah notifikasi.
        sinyalHubung.wait(gembok, [&] { return terhubung == 2; });
    };

    // 2) Thread penerima: accept DUA client lebih dulu, baru satu thread
    //    penerima per client dijalankan — daftar koneksi sudah lengkap
    //    sehingga tidak ada broadcast yang terlewat.
    std::thread threadPenerima([&fdListen, &registri]() {
        std::vector<int> fds;
        for (int i = 0; i < 2; ++i) {
            const int fdKoneksi = ::accept(fdListen, nullptr, nullptr);
            if (fdKoneksi < 0) {
                break;
            }
            fds.push_back(fdKoneksi);
            registri.tambah(fdKoneksi);
        }
        for (int fd : fds) {
            std::thread(layaniClient, fd, std::ref(registri)).detach();
        }
    });

    // 3) Dua client terhubung, saling berkirim, dan baru pamit setelah
    //    seluruh siaran yang diharapkan diterima (mencegah tutup lebih dulu).
    struct CatatanClient {
        std::vector<std::string> diterima;
    };
    CatatanClient catatan[2];
    const int targetDiterima[2] = {1, 2};   // Andi 1 siaran, Bela 2 siaran
    std::vector<std::thread> threadClient;
    int jumlahSelesai = 0;
    std::mutex kunciSelesai;
    std::condition_variable sinyalSelesai;

    auto buatClient = [&](const char* nama, int slot,
                          const std::vector<std::string>& pesanDikirim) {
        threadClient.emplace_back([&, nama, slot, pesanDikirim]() {
            const int fd = ::socket(AF_INET, SOCK_STREAM, 0);
            sockaddr_in tujuan{};
            tujuan.sin_family = AF_INET;
            tujuan.sin_port = port;
            tujuan.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
            if (::connect(fd, reinterpret_cast<const sockaddr*>(&tujuan),
                          sizeof(tujuan)) < 0) {
                return;
            }
            tungguBersama();   // mulai berkirim hanya setelah keduanya siap
            for (const std::string& teks : pesanDikirim) {
                kirimSemua(fd, std::string(nama) + "|" + teks + "\n");
            }
            // Menerima siaran sampai target terpenuhi — koneksi tetap
            // terbuka sehingga broadcast dari client lain tidak hilang.
            std::string baris;
            while (static_cast<int>(catatan[slot].diterima.size()) <
                   targetDiterima[slot]) {
                if (!terimaBaris(fd, baris)) {
                    break;
                }
                catatan[slot].diterima.push_back(baris);
            }
            kirimSemua(fd, "/quit\n");
            {
                std::lock_guard<std::mutex> gembok(kunciSelesai);
                ++jumlahSelesai;
            }
            sinyalSelesai.notify_all();
        });
    };

    buatClient("Andi", 0, {"Halo semua, selamat sore", "Siap latihan"});
    buatClient("Bela", 1, {"Halo Andi"});

    // 4) Menunggu kedua client selesai, lalu join semua thread.
    {
        std::unique_lock<std::mutex> gembok(kunciSelesai);
        sinyalSelesai.wait(gembok, [&] { return jumlahSelesai == 2; });
    }
    for (std::thread& t : threadClient) {
        t.join();
    }
    threadPenerima.join();

    // 5) Semua thread yang mencetak sudah berhenti: sekarang aman mencetak.
    std::cout << "[Andi] menerima " << catatan[0].diterima.size()
              << " pesan:\n";
    for (const std::string& baris : catatan[0].diterima) {
        std::cout << "        " << baris << "\n";
    }
    std::cout << "[Bela] menerima " << catatan[1].diterima.size()
              << " pesan:\n";
    for (const std::string& baris : catatan[1].diterima) {
        std::cout << "        " << baris << "\n";
    }

    // 6) Shutdown bersih: tutup pendengar, laporkan ringkasan.
    ::close(fdListen);
    std::cout << "[ringkasan] " << registri.siaran.size()
              << " pesan di-broadcast, 2 client selesai, server ditutup\n";
    return 0;
}

Output yang Dihasilkan:

text 
[server] siap di 127.0.0.1, port ephemeral dari getsockname
[Andi] menerima 1 pesan:
        Bela|Halo Andi
[Bela] menerima 2 pesan:
        Andi|Halo semua, selamat sore
        Andi|Siap latihan
[ringkasan] 3 pesan di-broadcast, 2 client selesai, server ditutup

Bedah Bagian yang Baru #

Mari kita telaah bagian yang menambah kemampuan server dibanding kerangka bagian 1:

  1. struct Registri: wadah bersama berisi daftar fds (koneksi aktif) dan siaran (pesan ter-broadcast). Setiap aksesnya dijaga std::mutex — thread penerima datang dari arah yang berbeda-beda, dan daftar yang bersama tanpa kunci berarti data race.
  2. layaniClient(fd, registri): badan thread penerima. Perulangan membaca baris satu per satu lewat terimaBaris, menolak baris perintah berawalan /, lalu meneruskan pesan sah ke registri.siarkan. Saat client mengirim /quit atau menutup koneksi, perulangan berakhir dan socket ditutup oleh thread yang sama — koneksi milik thread itu, rapinya juga oleh thread itu.
  3. siarkan(pesan, fdPengirim): mengirim pesan ke semua koneksi KECUALI pengirimnya, lalu mencatat pesan ke siaran. Pengirim tidak perlu menerima perkataannya sendiri — perilaku yang pembaca harapkan dari aplikasi chat.
  4. threadPenerima: menerima dua koneksi lebih dulu (fds dikumpulkan), baru menyalakan thread layaniClient. Inilah kunci determinisme skenario: tidak ada siaran yang terjadi ketika daftar belum lengkap.
  5. tungguBersama() (rendezvous client): kedua client menaikkan terhubung, lalu memanggil notify_all() dan menunggu predikat terhubung == 2. Client yang tiba kedua langsung lolos karena predikatnya sudah benar; client yang tiba pertama dibangunkan oleh notify_all. Pola increment-notify-wait ini menutup celah “notifikasi datang sebelum menunggu” yang menjadi penyebab deadlock klasik.
  6. Pencetakan di akhir: catatan[2] menyimpan hasil penerimaan per client; teks hanya dicetak oleh main setelah join — urutan keluaran tidak lagi bergantung pada jadwal thread mana pun.

Dengan server hidup dan dua client saling berkirim, fondasi chat sudah jalan. Masih ada satu kasarannya: bentuk pesan di kawat masih ditulis langsung di beberapa tempat. Bagian 3 memisahkannya menjadi protokol line-based dengan parser yang bisa diuji sendiri.

Rujukan resmi: man7 — pthread overview , cppreference — std::thread , cppreference — std::condition_variable .