// The Protomux adapter, end to end: two real Protomux instances wired to each
// other in memory, each driving a real Protocol, moving a real file.
//
// This is the first test where frames go through actual Protomux framing rather
// than being handed between objects, so it is the first that could catch a
// framing or slot-routing mistake. It still needs no socket — Protomux takes
// bytes in via recv() and hands bytes out via a write callback, so two muxes
// wire together directly, the same way the JS harness pairs two overlays over a
// duplex pair.
//
// Build (from hyperswarm-peer/):
// g++ -O2 -std=c++17 -I liboverlay/include -I src $(pkg-config --cflags libsodium) \
// -o /tmp/ho_adapter liboverlay/test/test_transport_adapter.cpp src/protomux.cpp \
// $(pkg-config --libs libsodium)
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>
#include "overlay_transport.h"
using namespace hyperoverlay;
using namespace hyperswarm;
static int passed = 0, failed = 0;
static void check(bool c, const std::string &what) {
if (c) { passed++; printf(" ok %s\n", what.c_str()); }
else { failed++; printf(" FAIL %s\n", what.c_str()); }
}
static std::vector<uint8_t> make_bytes(size_t n, uint32_t seed) {
std::vector<uint8_t> v(n);
uint32_t s = seed;
for (size_t i = 0; i < n; i++) { s = s * 1664525u + 1013904223u; v[i] = (uint8_t)(s >> 24); }
return v;
}
static std::vector<uint8_t> read_file_bytes(const std::string &path) {
std::vector<uint8_t> out;
FILE *f = std::fopen(path.c_str(), "rb");
if (!f) return out;
std::fseek(f, 0, SEEK_END);
const long n = std::ftell(f);
std::fseek(f, 0, SEEK_SET);
if (n > 0) { out.resize((size_t)n); if (std::fread(out.data(), 1, out.size(), f) != out.size()) out.clear(); }
std::fclose(f);
return out;
}
int main() {
printf("hyper-overlay C++ Protomux transport adapter\n");
const std::vector<uint8_t> content = make_bytes(200 * 1024, 21);
uint8_t oid_raw[32];
cdc::hash_chunk(content.data(), content.size(), oid_raw);
const std::string oid = to_hex(oid_raw, 32);
index::FileIndex ia, ib;
std::remove("/tmp/claude-1000/ho_ad_a.log");
std::remove("/tmp/claude-1000/ho_ad_b.log");
ia.open("/tmp/claude-1000/ho_ad_a.log");
ib.open("/tmp/claude-1000/ho_ad_b.log");
// The holder serves from a real file and the requester assembles into one —
// neither side holds the whole thing, here or in the game.
const std::string src = "/tmp/claude-1000/ho_ad_src.bin";
const std::string dest = "/tmp/claude-1000/ho_ad_dst.bin";
{
FILE *f = std::fopen(src.c_str(), "wb");
std::fwrite(content.data(), 1, content.size(), f);
std::fclose(f);
}
std::remove(dest.c_str());
std::remove(xfer::partial_path_for(dest).c_str());
Protocol alice(&ia, [&](const std::string &h, ContentLocation &out) {
if (h != oid) return false; out.path = src; return true;
});
Protocol bob(&ib, [](const std::string &, ContentLocation &) { return false; });
Protomux mux_a, mux_b;
// Wire the two muxes together. Bytes out of one are bytes into the other —
// queued rather than delivered inline, because a write that re-enters the
// peer's recv() during its own send would recurse through the whole transfer.
std::vector<std::pair<Protomux *, std::vector<uint8_t>>> wire;
size_t bytes_a_to_b = 0, bytes_b_to_a = 0;
// The write callbacks now answer whether the carrier can take more — the
// signal libudx reports and every layer used to discard. `a_capacity` lets the
// test wedge alice's side the way a full socket buffer would.
int a_capacity = -1, a_writes = 0;
mux_a.set_write_callback([&](const uint8_t *d, size_t n) {
bytes_a_to_b += n;
wire.push_back({ &mux_b, std::vector<uint8_t>(d, d + n) });
if (a_capacity < 0) return true;
return ++a_writes < a_capacity;
});
mux_b.set_write_callback([&](const uint8_t *d, size_t n) {
bytes_b_to_a += n;
wire.push_back({ &mux_a, std::vector<uint8_t>(d, d + n) });
return true;
});
OverlayTransport ta(&mux_a, &alice, "alice");
OverlayTransport tb(&mux_b, &bob, "bob");
alice.set_transport(&ta);
bob.set_transport(&tb);
check(ta.ok() && tb.ok(), "both channels created on protocol " + std::string(OVERLAY_PROTOCOL));
check(!ta.slot_mismatch() && !tb.slot_mismatch(),
"Protomux slot indices match registration order (MsgId is the wire order)");
auto pump = [&]() {
for (int guard = 0; guard < 100000 && !wire.empty(); guard++) {
auto item = wire.front();
wire.erase(wire.begin());
item.first->recv(item.second.data(), item.second.size());
}
};
pump(); // let the channel handshakes settle
bool done = false;
bob.on_complete = [&](const std::string &, const std::string &) { done = true; };
bob.request(oid, dest);
pump();
std::vector<uint8_t> received;
{
FILE *f = std::fopen(dest.c_str(), "rb");
if (f) {
std::fseek(f, 0, SEEK_END);
const long n = std::ftell(f);
std::fseek(f, 0, SEEK_SET);
if (n > 0) { received.resize((size_t)n); if (std::fread(received.data(), 1, received.size(), f) != received.size()) received.clear(); }
std::fclose(f);
}
}
check(done, "transfer completed through real Protomux framing");
check(received.size() == content.size(), "received length matches");
check(received == content, "bytes reconstruct the original exactly");
// Direction sanity: the holder should ship far more than the requester, since
// one carries the file and the other carries requests. If these were close,
// something is echoing.
check(bytes_a_to_b > bytes_b_to_a * 4,
"holder->requester carries the payload (" + std::to_string(bytes_a_to_b) +
" vs " + std::to_string(bytes_b_to_a) + " bytes)");
check(bytes_a_to_b >= content.size(),
"at least the file's worth of bytes crossed the wire");
// ---- backpressure, end to end through real Protomux --------------------
//
// Everything above this point could be satisfied by a carrier that always
// claims to be writable. This wedges the holder's side after a few writes and
// checks the pump actually stops — through real framing, a real channel, and
// the real adapter, rather than a stub that reports what the test wants.
{
const std::string dest2 = "/tmp/claude-1000/ho_ad_dst2.bin";
std::remove(dest2.c_str());
std::remove(xfer::partial_path_for(dest2).c_str());
a_capacity = 4; // alice's carrier fills after 4 writes
a_writes = 0;
const size_t before = bytes_a_to_b;
bool done2 = false;
bob.on_complete = [&](const std::string &, const std::string &) { done2 = true; };
bob.request(oid, dest2);
pump();
check(!done2, "with the holder's carrier wedged, the transfer does not complete");
check(bytes_a_to_b > before, "some bytes did move before it parked");
// Drain it, exactly as udx's on_drain would.
int rounds = 0;
while (!done2 && rounds++ < 200) {
a_writes = 0;
mux_a.notify_drain();
pump();
}
check(done2, "and it completes once the carrier drains");
check(read_file_bytes(dest2) == content, "with byte-identical content");
}
printf(" %d assertions pass, %d fail\n", passed, failed);
printf("%s\n", failed == 0 ? "PASS" : "FAIL");
return failed == 0 ? 0 : 1;
}