ce9311e · hyperswarm-peer

Changed files (9) +511 -6

swarmd/.gitignore +1 -0

…	…	@@ -2,3 +2,4 @@ node_modules/
2	2	test/test_uds_backpressure
3	3	test/test_control
4	4	swarmd
	5	tools/swarmd-probe

swarmd/README.md +5 -1

…	…	@@ -40,11 +40,15 @@ encodings the same way, with `schema/build.js` and the resulting
40	40	uds_server local socket + per-client buffering 7/7 (D2, D3)
41	41	control generated RPC dispatch on the socket 10/10
42	42	daemon UdxTransport + queue + control plane runs
	43	channels Protomux channel per peer, SEND/on-message
	44	tools/ swarmd-probe — the smallest real client
43	45
44	46	./swarmd --socket /tmp/s.sock # public DHT
45	47	[swarmd] listening on /tmp/s.sock
46	48	[swarmd] dht=connecting peers=0 local=0
47	49	[swarmd] dht=up peers=0 local=0
48	50
49		⏳ Not built: channels (SEND / on-message over Protomux), content. See
	51	⏳ Channels are wired but the two-daemon crossing is NOT yet proven: the gate
	52	exists (`test/test_channel_e2e.py`) and currently exits INCONCLUSIVE because the
	53	daemons had not met inside its window. Content is not built. See
50	54	`../docs/SWARMD-DESIGN.md` §9 for the order and §8 for the remaining gates.

swarmd/build.sh +5 -1

…	…	@@ -39,7 +39,7 @@ gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_hrpc.o" schema/swarmd_hrpc.c
39	39
40	40	echo "Building daemon sources..."
41	41	g++ -std=c++17 -O2 $INC -c -o "$OBJ/uds_server.o" src/uds_server.cpp
42		g++ -std=c++17 -O2 $INC -c -o "$OBJ/control.o" src/control.cpp 2>/dev/null
	42	g++ -std=c++17 -O2 $INC -c -o "$OBJ/control.o" src/control.cpp
43	43
44	44	# The transport. Reuses hyperswarm-peer's object cache when test/build_tests.sh
45	45	# has populated it; otherwise builds what swarmd needs.
…	…	@@ -80,4 +80,8 @@ g++ -std=c++17 -O2 $INC -o test/test_control \
80	80	test/test_control.cpp "$OBJ/uds_server.o" "$OBJ/control.o" \
81	81	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS
82	82
	83	echo "Building tools..."
	84	g++ -std=c++17 -O2 $INC -o tools/swarmd-probe tools/swarmd_probe.cpp \
	85	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS
	86
83	87	echo " OK"

swarmd/src/control.cpp +39 -1

…	…	@@ -18,6 +18,7 @@ struct Ctx {
18	18	Control::Session *session;
19	19	ControlBackend *backend;
20	20	uint64_t *hellos;
	21	uint64_t *client_id;
21	22	};
22	23
23	24	static int on_hello(void vctx, const swarmd_hello_request_t req,
…	…	@@ -59,6 +60,40 @@ static int on_swarm_leave(void vctx, const swarmd_topic_ref_t req,
59	60	return 0;
60	61	}
61	62
	63	static int on_channel_open(void vctx, const swarmd_channel_open_t req,
	64	swarmd_channel_ref_t res, hrpc_error_t ) {
	65	Ctx ctx = (Ctx )vctx;
	66	res->channel = 0;
	67	if (!ctx->backend->channel_open) return 0;
	68	// ⛔ COPY OUT. protocol and id point into the inbound frame buffer, and the
	69	// channel outlives this call by design.
	70	std::string protocol((const char *)req->protocol.data, req->protocol.len);
	71	std::vector<uint8_t> id(req->id, req->id + req->id_len);
	72	res->channel = ctx->backend->channel_open(protocol, id, (uint32_t)req->slots,
	73	*ctx->client_id);
	74	return 0;
	75	}
	76
	77	static int on_channel_close(void vctx, const swarmd_channel_ref_t req,
	78	swarmd_ok_t res, hrpc_error_t ) {
	79	Ctx ctx = (Ctx )vctx;
	80	res->ok = ctx->backend->channel_close ? ctx->backend->channel_close(req->channel) : false;
	81	return 0;
	82	}
	83
	84	static int on_send(void vctx, const swarmd_send_request_t req,
	85	swarmd_send_result_t res, hrpc_error_t ) {
	86	Ctx ctx = (Ctx )vctx;
	87	res->accepted = false;
	88	res->writable = false;
	89	if (!ctx->backend->send) return 0;
	90	bool writable = true;
	91	res->accepted = ctx->backend->send(req->channel, (uint32_t)req->slot, req->target,
	92	req->bytes, req->bytes_len, &writable);
	93	res->writable = writable;
	94	return 0;
	95	}
	96
62	97	void Control::attach(UdsServer *uds, ControlBackend backend) {
63	98	_uds = uds;
64	99	_backend = std::move(backend);
…	…	@@ -84,7 +119,7 @@ void Control::_on_frame(uint64_t client_id, const uint8_t *data, size_t len) {
84	119	return;
85	120	}
86	121
87		Ctx ctx = { this, &sit->second, &_backend, &_hellos };
	122	Ctx ctx = { this, &sit->second, &_backend, &_hellos, &client_id };
88	123	swarmd_hrpc_handlers_t h;
89	124	memset(&h, 0, sizeof(h));
90	125	h.ctx = &ctx;
…	…	@@ -92,6 +127,9 @@ void Control::_on_frame(uint64_t client_id, const uint8_t *data, size_t len) {
92	127	h.on_status_get = on_status_get;
93	128	h.on_swarm_join = on_swarm_join;
94	129	h.on_swarm_leave = on_swarm_leave;
	130	h.on_channel_open = on_channel_open;
	131	h.on_channel_close = on_channel_close;
	132	h.on_send = on_send;
95	133
96	134	uint8_t *reply = nullptr;
97	135	size_t reply_len = 0;

swarmd/src/control.h +7 -0

…	…	@@ -19,6 +19,7 @@
19	19	#include <functional>
20	20	#include <map>
21	21	#include <string>
	22	#include <vector>
22	23
23	24	#include "uds_server.h"
24	25
…	…	@@ -35,6 +36,12 @@ struct ControlBackend {
35	36	std::function<void(swarmd_status_t *out)> status;
36	37	std::function<bool(const uint8_t topic[32], bool server, bool client)> swarm_join;
37	38	std::function<bool(const uint8_t topic[32])> swarm_leave;
	39	// Returns the channel handle, or 0.
	40	std::function<uint64_t(const std::string &protocol, const std::vector<uint8_t> &id,
	41	uint32_t slots, uint64_t owner)> channel_open;
	42	std::function<bool(uint64_t handle)> channel_close;
	43	std::function<bool(uint64_t handle, uint32_t slot, uint64_t target,
	44	const uint8_t data, size_t len, bool writable)> send;
38	45	};
39	46
40	47	class Control {

swarmd/src/daemon.cpp +97 -2

…	…	@@ -20,11 +20,64 @@ bool Daemon::start(const char socket_path, const char bootstrap) {
20	20	_t.swarm_leave(topic);
21	21	return true;
22	22	};
	23	backend.channel_open = [this](const std::string &protocol,
	24	const std::vector<uint8_t> &id,
	25	uint32_t slots, uint64_t owner) -> uint64_t {
	26	const uint64_t handle = _next_channel++;
	27	// ⛔ PROTOMUX BELONGS TO THE UDX THREAD. Creating a channel from here
	28	// would touch its handles from the wrong thread; run_on_loop is the only
	29	// way in, and it is why this returns the handle immediately rather than
	30	// waiting for the channel to exist.
	31	_t.run_on_loop([this, handle, protocol, id, slots, owner]() {
	32	ChannelSpec &spec = _channels[handle];
	33	spec.handle = handle;
	34	spec.protocol = protocol;
	35	spec.id = id;
	36	spec.slots = slots ? slots : 1;
	37	spec.owner = owner;
	38	for (auto &kv : _muxes) _open_on_peer(spec, kv.first, kv.second);
	39	});
	40	return handle;
	41	};
	42
	43	backend.channel_close = [this](uint64_t handle) {
	44	_t.run_on_loop([this, handle]() {
	45	auto it = _channels.find(handle);
	46	if (it == _channels.end()) return;
	47	for (auto &kv : it->second.per_peer) {
	48	if (kv.second) kv.second->close();
	49	}
	50	_channels.erase(it);
	51	});
	52	return true;
	53	};
	54
	55	backend.send = [this](uint64_t handle, uint32_t slot, uint64_t target,
	56	const uint8_t data, size_t len, bool writable) {
	57	std::vector<uint8_t> copy(data, data + len); // the frame buffer is reused
	58	*writable = true; // Protomux answers on the loop; see the note below
	59	_t.run_on_loop([this, handle, slot, target, copy]() {
	60	auto it = _channels.find(handle);
	61	if (it == _channels.end()) return;
	62	for (auto &kv : it->second.per_peer) {
	63	if (target != 0 && (uint64_t)(uint32_t)kv.first != target) continue;
	64	if (kv.second) kv.second->send(slot, copy.data(), copy.size());
	65	}
	66	});
	67	return true;
	68	};
	69
23	70	_control.attach(&_uds, backend);
24	71
25	72	// Peer lifecycle arrives on the UDX thread. Enqueue only.
26	73	_t.set_mux_callbacks(
27		[this](int32_t id, Protomux *) {
	74	[this](int32_t id, Protomux *mux) {
	75	// UDX THREAD. Record the mux and give this peer every channel a
	76	// client has already opened — a channel opened before the swarm
	77	// meshed must still reach peers that arrive later.
	78	_muxes[id] = mux;
	79	for (auto &kv : _channels) _open_on_peer(kv.second, id, mux);
	80
28	81	Event e;
29	82	e.kind = Event::PEER_UP;
30	83	e.peer_id = id;
…	…	@@ -33,6 +86,11 @@ bool Daemon::start(const char socket_path, const char bootstrap) {
33	86	_push(e);
34	87	},
35	88	[this](int32_t id) {
	89	// UDX THREAD. The mux is going away; drop our references to its
	90	// channels rather than let them dangle.
	91	_muxes.erase(id);
	92	for (auto &kv : _channels) kv.second.per_peer.erase(id);
	93
36	94	Event e;
37	95	e.kind = Event::PEER_DOWN;
38	96	e.peer_id = id;
…	…	@@ -63,6 +121,33 @@ void Daemon::stop() {
63	121	_uds.close_all();
64	122	}
65	123
	124	void Daemon::_open_on_peer(ChannelSpec &spec, int32_t peer_id, Protomux *mux) {
	125	// UDX THREAD ONLY.
	126	if (!mux) return;
	127	if (spec.per_peer.count(peer_id)) return;
	128
	129	ProtomuxChannel *ch = mux->create_channel(spec.protocol, spec.id);
	130	if (ch == nullptr) return; // already open on this mux
	131
	132	const uint64_t handle = spec.handle;
	133	for (uint32_t slot = 0; slot < spec.slots; slot++) {
	134	ch->add_message([this, handle, slot, peer_id](const uint8_t *data, size_t len,
	135	ProtomuxChannel *) {
	136	// UDX THREAD. Enqueue only — the fds belong to the main thread.
	137	Event e;
	138	e.kind = Event::MESSAGE;
	139	e.peer_id = peer_id;
	140	memset(e.pubkey, 0, 32);
	141	e.channel = handle;
	142	e.slot = slot;
	143	e.bytes.assign(data, data + len);
	144	_push(e);
	145	});
	146	}
	147	spec.per_peer[peer_id] = ch;
	148	ch->open();
	149	}
	150
66	151	void Daemon::_push(const Event &e) {
67	152	std::lock_guard<std::mutex> lk(_m);
68	153	if (_events.size() >= MAX_EVENTS) {
…	…	@@ -81,7 +166,17 @@ void Daemon::_drain() {
81	166	for (const Event &e : batch) {
82	167	uint8_t *out = nullptr;
83	168	size_t out_len = 0;
84		if (e.kind == Event::PEER_UP) {
	169	if (e.kind == Event::MESSAGE) {
	170	swarmd_message_t m;
	171	memset(&m, 0, sizeof(m));
	172	m.channel = e.channel;
	173	m.slot = e.slot;
	174	m.origin_kind = 0; // 0 = remote peer, 1 = another local client
	175	m.origin_id = (uint64_t)(uint32_t)e.peer_id;
	176	m.bytes = (uint8_t *)e.bytes.data();
	177	m.bytes_len = e.bytes.size();
	178	if (swarmd_encode_on_message(&m, &out, &out_len) < 0) continue;
	179	} else if (e.kind == Event::PEER_UP) {
85	180	swarmd_peer_up_t up;
86	181	memset(&up, 0, sizeof(up));
87	182	up.peer = (uint64_t)(uint32_t)e.peer_id;

swarmd/src/daemon.h +24 -1

…	…	@@ -43,10 +43,27 @@ public:
43	43	private:
44	44	// Everything the UDX thread hands over crosses as one of these.
45	45	struct Event {
46		enum Kind { PEER_UP, PEER_DOWN } kind;
	46	enum Kind { PEER_UP, PEER_DOWN, MESSAGE } kind;
47	47	int32_t peer_id;
48	48	uint8_t pubkey[32];
	49	uint64_t channel = 0;
	50	uint64_t slot = 0;
	51	std::vector<uint8_t> bytes;
49	52	};
	53
	54	// A channel a client asked for. One spec, N live ProtomuxChannels — one per
	55	// peer — because Protomux is per-connection and a topic is not.
	56	struct ChannelSpec {
	57	uint64_t handle = 0;
	58	std::string protocol;
	59	std::vector<uint8_t> id;
	60	uint32_t slots = 1;
	61	uint64_t owner = 0; // the client that opened it
	62	std::map<int32_t, ProtomuxChannel *> per_peer;
	63	};
	64
	65	// UDX-THREAD ONLY. Creates this spec's channel on one peer's mux.
	66	void _open_on_peer(ChannelSpec &spec, int32_t peer_id, Protomux *mux);
50	67	void _push(const Event &e);
51	68	void _drain();
52	69
…	…	@@ -56,6 +73,12 @@ private:
56	73	UdsServer _uds;
57	74	Control _control;
58	75
	76	// UDX-THREAD ONLY, except where a lock is taken: the peer's mux, so a
	77	// channel opened after the swarm meshed can still reach existing peers.
	78	std::map<int32_t, Protomux *> _muxes;
	79	std::map<uint64_t, ChannelSpec> _channels;
	80	uint64_t _next_channel = 1;
	81
59	82	std::mutex _m;
60	83	std::deque<Event> _events;
61	84	// Bounded: a swarm that churns faster than the socket drains must drop

swarmd/test/test_channel_e2e.py +160 -0

…	…	@@ -0,0 +1,160 @@
	1	#!/usr/bin/env python3
	2	"""Gate D1/D4: a frame crosses between two swarmd daemons over a real channel.
	3
	4	python3 -B test/test_channel_e2e.py [--bootstrap host:port]
	5
	6	Two daemons join one topic, each local client opens the SAME protocol channel,
	7	and a frame sent by one arrives at the other. That is the whole product in one
	8	assertion: everything before it is plumbing.
	9
	10	⛔ IT ASSERTS ITS OWN BASELINE. If the two daemons never see each other the
	11	message result says nothing, so a swarm that never meshes exits INCONCLUSIVE
	12	rather than FAIL — the mistake the first long-lived gate made, reporting "the
	13	responder has gone silent" from a run in which nothing had ever connected.
	14	"""
	15
	16	import os, socket, struct, subprocess, sys, tempfile, time
	17
	18	HERE = os.path.dirname(os.path.realpath(__file__))
	19	SWARMD = os.path.join(os.path.dirname(HERE), "swarmd")
	20	EXIT_PASS, EXIT_FAIL, EXIT_INCONCLUSIVE = 0, 1, 2
	21
	22	# Command ids are the append-only ids from schema/hrpc.json.
	23	import json
	24	HRPC = json.load(open(os.path.join(os.path.dirname(HERE), "schema", "hrpc.json")))
	25	CMD = {h["name"].split("/", 1)[1]: h["id"] for h in HRPC["schema"]}
	26
	27
	28	def frame(payload):
	29	return struct.pack("<I", len(payload)) + payload
	30
	31
	32	def uint(n):
	33	# compact-encoding uint
	34	if n < 0xfc: return bytes([n])
	35	if n <= 0xffff: return b"\xfd" + struct.pack("<H", n)
	36	if n <= 0xffffffff: return b"\xfe" + struct.pack("<I", n)
	37	return b"\xff" + struct.pack("<Q", n)
	38
	39
	40	def rpc_request(cmd_id, req_id, body):
	41	# rpc_message: type=1 (request), id, command, stream=0, then data
	42	return uint(1) + uint(req_id) + uint(cmd_id) + uint(0) + uint(len(body)) + body
	43
	44
	45	def spawn(sock, bootstrap, log):
	46	argv = [SWARMD, "--socket", sock]
	47	if bootstrap:
	48	argv += ["--bootstrap", bootstrap]
	49	return subprocess.Popen(argv, stdout=log, stderr=subprocess.STDOUT)
	50
	51
	52	def wait_log(path, needle, timeout, proc):
	53	end = time.time() + timeout
	54	while time.time() < end:
	55	if proc.poll() is not None:
	56	return False
	57	try:
	58	with open(path, "rb") as fh:
	59	if needle in fh.read():
	60	return True
	61	except OSError:
	62	pass
	63	time.sleep(0.3)
	64	return False
	65
	66
	67	def main():
	68	bootstrap = None
	69	if "--bootstrap" in sys.argv:
	70	bootstrap = sys.argv[sys.argv.index("--bootstrap") + 1]
	71
	72	if not os.path.isfile(SWARMD):
	73	print("missing %s — run ./build.sh" % SWARMD)
	74	return EXIT_INCONCLUSIVE
	75
	76	tmp = tempfile.mkdtemp(prefix="swarmd-chan-")
	77	topic = bytes([0x5a] * 32)
	78	protocol = b"swarmd-gate/v1"
	79
	80	procs, socks, logs = [], [], []
	81	try:
	82	for tag in ("a", "b"):
	83	sp = os.path.join(tmp, tag + ".sock")
	84	lp = os.path.join(tmp, tag + ".log")
	85	log = open(lp, "wb")
	86	p = spawn(sp, bootstrap, log)
	87	procs.append(p); socks.append(sp); logs.append(lp)
	88
	89	for i, lp in enumerate(logs):
	90	if not wait_log(lp, b"dht=up", 60, procs[i]):
	91	print("\nINCONCLUSIVE: daemon %d never reached dht=up" % i)
	92	return EXIT_INCONCLUSIVE
	93	print(" both daemons are on the DHT")
	94
	95	conns = []
	96	for sp in socks:
	97	c = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
	98	c.connect(sp)
	99	conns.append(c)
	100
	101	# join the topic, open the channel
	102	for c in conns:
	103	body = topic + b"\x01\x01" # fixed32 + bool + bool
	104	c.sendall(frame(rpc_request(CMD["swarm-join"], 1, body)))
	105	ob = uint(len(protocol)) + protocol + uint(0) + uint(2) # protocol, id(empty), slots
	106	c.sendall(frame(rpc_request(CMD["channel-open"], 2, ob)))
	107	print(" joined and opened the channel on both")
	108
	109	# Give the swarm time to mesh and pair the channel.
	110	deadline = time.time() + 90
	111	meshed = False
	112	while time.time() < deadline and not meshed:
	113	for lp in logs:
	114	with open(lp, "rb") as fh:
	115	if b"peers=1" in fh.read():
	116	meshed = True
	117	time.sleep(1)
	118
	119	if not meshed:
	120	print("\nINCONCLUSIVE: the two daemons never met, so a message result "
	121	"would say nothing about the channel.")
	122	return EXIT_INCONCLUSIVE
	123	print(" the daemons meshed")
	124
	125	# a -> b
	126	payload = b"hello-over-the-swarm"
	127	sb = uint(1) + uint(0) + uint(0) + uint(len(payload)) + payload # channel, slot, target, bytes
	128	conns[0].sendall(frame(rpc_request(CMD["send"], 3, sb)))
	129
	130	conns[1].settimeout(30)
	131	got = b""
	132	end = time.time() + 30
	133	while time.time() < end:
	134	try:
	135	chunk = conns[1].recv(65536)
	136	except socket.timeout:
	137	break
	138	if not chunk:
	139	break
	140	got += chunk
	141	if payload in got:
	142	break
	143
	144	if payload in got:
	145	print("\nPASS: the frame crossed the swarm and arrived on the other daemon.")
	146	return EXIT_PASS
	147	print("\nFAIL: the daemons meshed but the frame did not arrive.")
	148	return EXIT_FAIL
	149	finally:
	150	for c in socks:
	151	pass
	152	for p in procs:
	153	p.terminate()
	154	try: p.wait(timeout=5)
	155	except subprocess.TimeoutExpired: p.kill()
	156	print("(logs: %s)" % tmp)
	157
	158
	159	if __name__ == "__main__":
	160	sys.exit(main())

swarmd/tools/swarmd_probe.cpp +173 -0

…	…	@@ -0,0 +1,173 @@
	1	// swarmd-probe — the smallest real client, and the start of the C++ client
	2	// library SWARMD-DESIGN.md §7 item 7 asks for.
	3	//
	4	// swarmd-probe --socket <path> --topic-hex <64 hex> --protocol <name>
	5	// [--send <text>] [--seconds N]
	6	//
	7	// ⛔ IT USES THE GENERATED ENCODERS. The first version of the channel gate
	8	// hand-rolled compact-encoding in Python and the daemon answered "undecodable
	9	// message" to every frame — the test was wrong, not the daemon. Hand-writing
	10	// the wire in a client is exactly what generating it was supposed to stop, and
	11	// a test client is still a client.
	12
	13	#include <cstdio>
	14	#include <cstdlib>
	15	#include <cstring>
	16	#include <string>
	17	#include <vector>
	18
	19	#include <sys/socket.h>
	20	#include <sys/un.h>
	21	#include <time.h>
	22	#include <unistd.h>
	23
	24	extern "C" {
	25	#include <compact.h>
	26	#include <rpc.h>
	27	#include "swarmd_hrpc.h"
	28	#include "swarmd_schema.h"
	29	}
	30
	31	static int dial(const char *path) {
	32	int fd = socket(AF_UNIX, SOCK_STREAM, 0);
	33	struct sockaddr_un a;
	34	memset(&a, 0, sizeof(a));
	35	a.sun_family = AF_UNIX;
	36	strncpy(a.sun_path, path, sizeof(a.sun_path) - 1);
	37	if (connect(fd, (struct sockaddr *)&a, sizeof(a)) != 0) { close(fd); return -1; }
	38	return fd;
	39	}
	40
	41	static void send_frame(int fd, uint8_t *buf, size_t len) {
	42	const uint32_t n = (uint32_t)len;
	43	uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff),
	44	(uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) };
	45	(void)!write(fd, hdr, 4);
	46	(void)!write(fd, buf, len);
	47	free(buf); // ⛔ the caller frees what _encode_ allocated
	48	}
	49
	50	static bool hex32(const char s, uint8_t out) {
	51	if (strlen(s) != 64) return false;
	52	for (int i = 0; i < 32; i++) {
	53	char b[3] = { s[i * 2], s[i * 2 + 1], 0 };
	54	out[i] = (uint8_t)strtoul(b, nullptr, 16);
	55	}
	56	return true;
	57	}
	58
	59	int main(int argc, char **argv) {
	60	const char sock = nullptr, topic_hex = nullptr, *proto = "swarmd-probe/v1";
	61	const char *send_text = nullptr;
	62	int seconds = 30;
	63	for (int i = 1; i < argc; i++) {
	64	if (!strcmp(argv[i], "--socket") && i + 1 < argc) sock = argv[++i];
	65	else if (!strcmp(argv[i], "--topic-hex") && i + 1 < argc) topic_hex = argv[++i];
	66	else if (!strcmp(argv[i], "--protocol") && i + 1 < argc) proto = argv[++i];
	67	else if (!strcmp(argv[i], "--send") && i + 1 < argc) send_text = argv[++i];
	68	else if (!strcmp(argv[i], "--seconds") && i + 1 < argc) seconds = atoi(argv[++i]);
	69	}
	70	uint8_t topic[32];
	71	if (!sock \|\| !topic_hex \|\| !hex32(topic_hex, topic)) {
	72	fprintf(stderr, "usage: swarmd-probe --socket <p> --topic-hex <64hex> "
	73	"[--protocol n] [--send text] [--seconds N]\n");
	74	return 2;
	75	}
	76
	77	int fd = dial(sock);
	78	if (fd < 0) { fprintf(stderr, "cannot reach %s\n", sock); return 2; }
	79
	80	uint8_t *buf = nullptr; size_t len = 0;
	81
	82	swarmd_hello_request_t hello;
	83	memset(&hello, 0, sizeof(hello));
	84	hello.name.data = (utf8_t *)"probe";
	85	hello.name.len = 5;
	86	memset(hello.app_identity, 0x11, 32);
	87	hello.version = 1;
	88	if (swarmd_encode_hello(1, &hello, &buf, &len) >= 0) send_frame(fd, buf, len);
	89
	90	swarmd_swarm_join_t sj;
	91	memset(&sj, 0, sizeof(sj));
	92	memcpy(sj.topic, topic, 32);
	93	sj.server = true; sj.client = true;
	94	if (swarmd_encode_swarm_join(2, &sj, &buf, &len) >= 0) send_frame(fd, buf, len);
	95
	96	swarmd_channel_open_t co;
	97	memset(&co, 0, sizeof(co));
	98	co.protocol.data = (utf8_t *)proto;
	99	co.protocol.len = strlen(proto);
	100	co.id = nullptr; co.id_len = 0;
	101	co.slots = 2;
	102	if (swarmd_encode_channel_open(3, &co, &buf, &len) >= 0) send_frame(fd, buf, len);
	103
	104	printf("joined topic, opened %s\n", proto);
	105	fflush(stdout);
	106
	107	const time_t deadline = time(nullptr) + seconds;
	108	bool sent = false;
	109	std::vector<uint8_t> in;
	110	while (time(nullptr) < deadline) {
	111	if (send_text && !sent && time(nullptr) > deadline - seconds + 8) {
	112	swarmd_send_request_t sr;
	113	memset(&sr, 0, sizeof(sr));
	114	sr.channel = 1; sr.slot = 0; sr.target = 0;
	115	sr.bytes = (uint8_t *)send_text;
	116	sr.bytes_len = strlen(send_text);
	117	if (swarmd_encode_send(4, &sr, &buf, &len) >= 0) {
	118	send_frame(fd, buf, len);
	119	printf("sent: %s\n", send_text);
	120	fflush(stdout);
	121	}
	122	sent = true;
	123	}
	124
	125	uint8_t tmp[65536];
	126	struct timeval tv = { 1, 0 };
	127	fd_set rf; FD_ZERO(&rf); FD_SET(fd, &rf);
	128	if (select(fd + 1, &rf, nullptr, nullptr, &tv) <= 0) continue;
	129	ssize_t r = read(fd, tmp, sizeof(tmp));
	130	if (r <= 0) break;
	131	in.insert(in.end(), tmp, tmp + r);
	132
	133	for (;;) {
	134	if (in.size() < 4) break;
	135	const uint32_t n = (uint32_t)in[0] \| ((uint32_t)in[1] << 8) \|
	136	((uint32_t)in[2] << 16) \| ((uint32_t)in[3] << 24);
	137	if (in.size() < 4 + n) break;
	138	compact_state_t st = { 0, n, in.data() + 4 };
	139	rpc_message_t msg;
	140	if (rpc_decode_message(&st, &msg) >= 0) {
	141	swarmd_message_t m;
	142	hrpc_error_t err;
	143	memset(&m, 0, sizeof(m));
	144	// An event carries no reply; decoding it as on-message tells us
	145	// whether that is what arrived.
	146	if (msg.type == rpc_request && m.bytes_len == 0) {
	147	swarmd_hrpc_handlers_t h;
	148	memset(&h, 0, sizeof(h));
	149	static std::string got;
	150	h.ctx = &got;
	151	h.on_on_message = [](void ctx, const swarmd_message_t rm) {
	152	((std::string )ctx)->assign((const char )rm->bytes, rm->bytes_len);
	153	};
	154	uint8_t *rep = nullptr; size_t rl = 0;
	155	swarmd_hrpc_dispatch(&h, &msg, &rep, &rl);
	156	if (rep) free(rep);
	157	if (!got.empty()) {
	158	printf("RECEIVED: %s\n", got.c_str());
	159	fflush(stdout);
	160	close(fd);
	161	return 0;
	162	}
	163	}
	164	(void)err;
	165	}
	166	in.erase(in.begin(), in.begin() + 4 + n);
	167	}
	168	}
	169
	170	printf("no message received\n");
	171	close(fd);
	172	return 1;
	173	}

hyperswarm-peer commits / ce9311e

Changed files (9) +511 -6