// Gate for UdxTransport's stream consumer (set_stream_consumer / send_frame /
// close_stream): the seam keetd uses to run its own protomux over the secret-stream.
//
// Over the DHT path (the production one), against a local JS testnet:
// node ../integration/layer2_testnet.js & # writes /tmp/layer_bootstrap.txt
// ./test_stream_consumer [host:port]
// Two transports on one topic, BOTH with a consumer:
// - each sees exactly one open; the initiator flags are opposite, the handshake
// hashes equal, and each side's remote key is the other's
// - frames arrive intact and in order both ways, including one of 60 KiB
// - NO internal Protomux: on_mux_ready never fires
// - close_stream on one side closes the stream, and `close` fires
// And a pair WITHOUT a consumer: the default path is unchanged — on_mux_ready fires
// (the godot-rpc channel opened), and no consumer callback does.
#include "noise_ik.h"
#include "udx_transport.h"
#include <sodium.h>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstring>
#include <fstream>
#include <string>
#include <unistd.h>
#include <mutex>
#include <thread>
#include <vector>
static int passed = 0, failed = 0;
static void check (bool ok, const char *what) {
(ok ? passed : failed)++;
printf(" %s %s\n", ok ? "PASS" : "FAIL", what);
}
static void sleep_ms (int ms) { std::this_thread::sleep_for(std::chrono::milliseconds(ms)); }
template <typename F> static bool wait_for (F f, int ms) {
for (int i = 0; i < ms / 10; i++) { if (f()) return true; sleep_ms(10); }
return f();
}
struct Side {
std::mutex m;
std::vector<UdxTransport::StreamInfo> opens;
std::vector<std::vector<uint8_t>> frames;
std::atomic<int> closes{ 0 }, muxReady{ 0 };
void attach (UdxTransport &t) {
t.set_stream_consumer(
[this] (const UdxTransport::StreamInfo &i) { std::lock_guard<std::mutex> l(m); opens.push_back(i); },
[this] (int32_t, const uint8_t *p, size_t n) { std::lock_guard<std::mutex> l(m); frames.emplace_back(p, p + n); },
[] (int32_t) {},
[this] (int32_t) { closes++; });
t.set_mux_callbacks([this] (int32_t, Protomux *) { muxReady++; }, nullptr);
}
size_t nopens () { std::lock_guard<std::mutex> l(m); return opens.size(); }
size_t nframes () { std::lock_guard<std::mutex> l(m); return frames.size(); }
};
static std::vector<uint8_t> pattern (size_t n, uint8_t seed) {
std::vector<uint8_t> v(n);
for (size_t i = 0; i < n; i++) v[i] = uint8_t(seed + i * 31);
return v;
}
static std::string g_bootstrap;
// two transports on the testnet, joined to one fresh topic
static bool mesh (UdxTransport &a, UdxTransport &b, Ed25519KeyPair &ka, Ed25519KeyPair &kb, const char *label) {
uint8_t topic[32];
char seed[80];
snprintf(seed, sizeof(seed), "stream-consumer/%s/%d", label, (int)getpid());
crypto_generichash(topic, 32, (const uint8_t *)seed, strlen(seed), nullptr, 0);
const char *bs[1] = { g_bootstrap.c_str() };
ka = NoiseIK::generate_keypair();
kb = NoiseIK::generate_keypair();
a.start_dht(ka, bs, 1);
b.start_dht(kb, bs, 1);
if (!wait_for([&] { return a.is_dht_active() && b.is_dht_active(); }, 15000)) return false;
a.swarm_join(topic, true, true);
b.swarm_join(topic, true, true);
return true;
}
static void test_consumer_pair () {
printf("\n--- stream consumer: two transports over the DHT ---\n");
UdxTransport a, b;
Side s, c;
s.attach(a);
c.attach(b);
Ed25519KeyPair ka, kb;
check(mesh(a, b, ka, kb, "consumer"), "both DHTs active, topic joined");
const bool opened = wait_for([&] { return s.nopens() == 1 && c.nopens() == 1; }, 30000);
check(opened, "each side saw exactly one open");
if (!opened) { printf(" opens: A=%zu B=%zu\n", s.nopens(), c.nopens()); a.stop(); b.stop(); return; }
UdxTransport::StreamInfo si, ci;
{ std::lock_guard<std::mutex> l(s.m); si = s.opens[0]; }
{ std::lock_guard<std::mutex> l(c.m); ci = c.opens[0]; }
check(si.is_initiator != ci.is_initiator, "initiator flags are opposite");
check(memcmp(si.handshake_hash, ci.handshake_hash, 64) == 0, "handshake hashes are equal");
bool nonzero = false;
for (int i = 0; i < 64; i++) nonzero |= si.handshake_hash[i] != 0;
check(nonzero, "handshake hash is not zeros");
check(memcmp(si.remote_public_key, kb.public_key, 32) == 0 && memcmp(ci.remote_public_key, ka.public_key, 32) == 0,
"each side's remote key is the other's");
// frames both ways, in order, one of 60 KiB
const std::vector<std::vector<uint8_t>> toA = { pattern(1, 1), pattern(300, 2), pattern(60 * 1024, 3), pattern(7, 4) };
const std::vector<std::vector<uint8_t>> toB = { pattern(5, 9), pattern(1000, 8) };
b.run_on_loop([&] { for (auto &f : toA) b.send_frame(ci.peer_id, f.data(), f.size()); });
a.run_on_loop([&] { for (auto &f : toB) a.send_frame(si.peer_id, f.data(), f.size()); });
const bool got = wait_for([&] { return s.nframes() == toA.size() && c.nframes() == toB.size(); }, 10000);
check(got, "every frame arrived");
{ std::lock_guard<std::mutex> l(s.m); check(s.frames == toA, "A received B's frames intact and in order"); }
{ std::lock_guard<std::mutex> l(c.m); check(c.frames == toB, "B received A's frames intact and in order"); }
check(s.muxReady == 0 && c.muxReady == 0, "no internal Protomux (on_mux_ready never fired)");
// close from B: both see the stream close
b.run_on_loop([&] { b.close_stream(ci.peer_id); });
check(wait_for([&] { return c.closes >= 1; }, 5000), "close_stream fires close locally");
check(wait_for([&] { return s.closes >= 1; }, 30000), "the remote sees the stream close");
// send_frame to a peer that is gone answers false
std::atomic<int> after{ -1 };
b.run_on_loop([&] { uint8_t x = 1; after = b.send_frame(ci.peer_id, &x, 1) ? 1 : 0; });
check(wait_for([&] { return after != -1; }, 2000) && after == 0, "send_frame to a closed peer answers false");
a.stop();
b.stop();
}
static void test_default_path_unchanged () {
printf("\n--- no consumer: the default path is unchanged ---\n");
UdxTransport a, b;
std::atomic<int> aReady{ 0 }, bReady{ 0 };
a.set_mux_callbacks([&] (int32_t, Protomux *) { aReady++; }, nullptr);
b.set_mux_callbacks([&] (int32_t, Protomux *) { bReady++; }, nullptr);
Ed25519KeyPair ka, kb;
check(mesh(a, b, ka, kb, "default"), "both DHTs active, topic joined");
check(wait_for([&] { return aReady >= 1 && bReady >= 1; }, 30000), "on_mux_ready fired on both sides (godot-rpc opened)");
a.stop();
b.stop();
}
int main (int argc, char **argv) {
if (sodium_init() < 0) return 2;
if (argc > 1) g_bootstrap = argv[1];
else { std::ifstream f("/tmp/layer_bootstrap.txt"); if (f) std::getline(f, g_bootstrap); }
const size_t comma = g_bootstrap.find(',');
if (comma != std::string::npos) g_bootstrap = g_bootstrap.substr(0, comma);
while (!g_bootstrap.empty() && isspace((unsigned char)g_bootstrap.back())) g_bootstrap.pop_back();
if (g_bootstrap.empty()) { printf("FAIL: no bootstrap. Start the testnet: node ../integration/layer2_testnet.js &\n"); return 1; }
test_consumer_pair();
test_default_path_unchanged();
printf("\n%s test_stream_consumer: %d/%d\n", failed ? "FAIL" : "PASS", passed, passed + failed);
return failed ? 1 : 0;
}