Skip to content

Commit ac76783

Browse files
gmelodiegmelodie
committed
chore(cbind): pubsub (nim-ffi part 5)
1 parent a485695 commit ac76783

4 files changed

Lines changed: 170 additions & 2 deletions

File tree

‎cbind/cbind.nimble‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,7 @@ task examples, "Build and run the C bindings examples":
149149
cborObjs.add obj
150150
let cborObjsStr = cborObjs.join(" ")
151151

152-
for example in ["echo"]:
152+
for example in ["echo", "gossipsub"]:
153153
let outBin = "../build/" & example
154154
exec "gcc -std=c11 -O2 -I c_bindings -I " & vendor & " examples/" & example & ".c " &
155155
cborObjsStr & " " & lib & " -pthread -Wl,-rpath,'$ORIGIN' -o " & outBin

‎cbind/examples/CMakeLists.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ set_property(TARGET tinycbor PROPERTY POSITION_INDEPENDENT_CODE ON)
6262

6363
find_package(Threads REQUIRED)
6464

65-
foreach(example echo)
65+
foreach(example echo gossipsub)
6666
add_executable(${example} "${example}.c")
6767
target_include_directories(${example} PRIVATE "${GEN_DIR}")
6868
target_link_libraries(${example} PRIVATE "${LIB_FILE}" tinycbor Threads::Threads)

‎cbind/examples/gossipsub.c‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
// GossipSub pub/sub: two TCP nodes connect on a shared topic and exchange one
2+
// message, delivered to the subscriber via its pubsub-message listener. The
3+
// generated bindings are asynchronous, so each call is wrapped with the
4+
// blocking helpers in common.h.
5+
#include "common.h"
6+
7+
// TransportType / MuxerType ordinals, mirrored from cbind/libp2p.nim.
8+
static const int64_t TransportTcp = 1;
9+
static const int64_t MuxerMplex = 0;
10+
11+
static const char* Topic = "/cbind/demo";
12+
13+
static atomic_int g_got = 0;
14+
static char g_received[512];
15+
16+
// gossipsub_publish replies with a PublishResponse carrying the number of peers
17+
// the message was forwarded to, so it needs its own waiter rather than on_bool.
18+
typedef struct {
19+
atomic_int done;
20+
int err_code;
21+
char err[256];
22+
int64_t peerCount;
23+
} PublishWaiter;
24+
25+
static void on_publish(int ec, const PublishResponse* reply, const char* em, void* ud) {
26+
PublishWaiter* w = (PublishWaiter*)ud;
27+
w->err_code = ec;
28+
if (reply)
29+
w->peerCount = reply->peerCount;
30+
if (em)
31+
snprintf(w->err, sizeof(w->err), "%s", em);
32+
atomic_store(&w->done, 1);
33+
}
34+
35+
static void onPubsubMessage(const PubsubMessageEvent* evt, void* ud) {
36+
(void)ud;
37+
size_t len = evt->data.len < sizeof(g_received) - 1 ? evt->data.len
38+
: sizeof(g_received) - 1;
39+
if (evt->data.data)
40+
memcpy(g_received, evt->data.data, len);
41+
g_received[len] = '\0';
42+
atomic_store(&g_got, 1);
43+
}
44+
45+
static LibP2PCtx* gossipsubNode(const char* listenAddr, const char* label) {
46+
NimFfiStr addrSlot = nimffi_str(listenAddr);
47+
Libp2pConfig cfg;
48+
memset(&cfg, 0, sizeof(cfg));
49+
cfg.mountGossipsub = true;
50+
cfg.gossipsubTriggerSelf = true;
51+
cfg.addrs.data = &addrSlot;
52+
cfg.addrs.len = 1;
53+
cfg.muxer = MuxerMplex;
54+
cfg.transport = TransportTcp;
55+
return await_create(&cfg, label);
56+
}
57+
58+
int main(void) {
59+
LibP2PCtx* subscriber = gossipsubNode("/ip4/127.0.0.1/tcp/5021", "subscriber");
60+
LibP2PCtx* publisher = gossipsubNode("/ip4/127.0.0.1/tcp/5022", "publisher");
61+
if (!subscriber || !publisher) {
62+
libp2p_ctx_destroy(subscriber);
63+
libp2p_ctx_destroy(publisher);
64+
return 1;
65+
}
66+
67+
int status = 1;
68+
BoolWaiter bw;
69+
70+
libp2p_ctx_add_on_pubsub_message_listener(subscriber, onPubsubMessage, NULL);
71+
72+
// Both peers subscribe so a GossipSub mesh can form between them.
73+
LibP2PCtx* nodes[2] = {subscriber, publisher};
74+
for (int i = 0; i < 2; i++) {
75+
if (!AWAIT_BOOL(bw,
76+
libp2p_ctx_gossipsub_subscribe(nodes[i], nimffi_str(Topic), on_bool,
77+
&bw),
78+
"subscribe") ||
79+
!AWAIT_BOOL(bw, libp2p_ctx_start(nodes[i], on_bool, &bw), "start"))
80+
goto cleanup;
81+
}
82+
83+
PeerInfoWaiter pw;
84+
if (!await_peerinfo(subscriber, &pw, "peerinfo"))
85+
goto cleanup;
86+
printf("Subscriber: %s\n", pw.peerId);
87+
88+
NimFfiStr connAddrs[16];
89+
for (size_t i = 0; i < pw.naddrs; i++)
90+
connAddrs[i] = nimffi_str(pw.addrs[i]);
91+
ConnectRequest connReq = {nimffi_str(pw.peerId), {connAddrs, pw.naddrs}, 0};
92+
if (!AWAIT_BOOL(bw, libp2p_ctx_connect(publisher, &connReq, on_bool, &bw), "connect"))
93+
goto cleanup;
94+
95+
// Give the subscription to propagate and the mesh to graft before publishing.
96+
sleep_ms(2000);
97+
98+
const char* message = "hello gossipsub";
99+
printf("Publishing: %s\n", message);
100+
NimFfiBytes payload = {(uint8_t*)message, strlen(message)};
101+
PublishRequest pubReq = {nimffi_str(Topic), payload};
102+
PublishWaiter pubw;
103+
memset(&pubw, 0, sizeof(pubw));
104+
libp2p_ctx_gossipsub_publish(publisher, &pubReq, on_publish, &pubw);
105+
if (!wait_done(&pubw.done) || pubw.err_code != 0) {
106+
fprintf(stderr, "publish: %s\n", pubw.err[0] ? pubw.err : "unknown");
107+
goto cleanup;
108+
}
109+
printf("Published to %lld peer(s)\n", (long long)pubw.peerCount);
110+
111+
if (wait_done(&g_got)) {
112+
printf("Subscriber received: %s\n", g_received);
113+
status = strcmp(g_received, message) == 0 ? 0 : 1;
114+
} else {
115+
fprintf(stderr, "Error: timed out waiting for the message\n");
116+
}
117+
118+
cleanup:
119+
AWAIT_BOOL(bw, libp2p_ctx_stop(publisher, on_bool, &bw), "stop publisher");
120+
AWAIT_BOOL(bw, libp2p_ctx_stop(subscriber, on_bool, &bw), "stop subscriber");
121+
libp2p_ctx_destroy(publisher);
122+
libp2p_ctx_destroy(subscriber);
123+
return status;
124+
}

‎cbind/libp2p_ffi.nim‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,15 +93,28 @@ type StreamReadLpRequest {.ffi.} = object
9393
streamId: uint64
9494
maxSize: int64
9595

96+
type PublishRequest {.ffi.} = object
97+
topic: string
98+
data: seq[byte]
99+
100+
type PublishResponse {.ffi.} = object
101+
peerCount: int ## number of peers the message was forwarded to
102+
96103
type ReadResponse {.ffi.} = object
97104
data: seq[byte]
98105

99106
type IncomingStreamEvent {.ffi.} = object
100107
proto: string
101108
streamId: uint64
102109

110+
type PubsubMessageEvent {.ffi.} = object
111+
topic: string
112+
data: seq[byte]
113+
103114
proc onIncomingStream*(event: IncomingStreamEvent) {.ffiEvent: "on_incoming_stream".}
104115

116+
proc onPubsubMessage*(event: PubsubMessageEvent) {.ffiEvent: "on_pubsub_message".}
117+
105118
proc register(reg: StreamRegistry, stream: Stream): uint64 =
106119
reg.nextStreamId.inc()
107120
let id = reg.nextStreamId
@@ -558,4 +571,35 @@ proc libp2pMountProtocol*(
558571
lib.customProtocols[proto] = mountedProtocol
559572
ok(true)
560573

574+
proc libp2pGossipsubPublish*(
575+
lib: LibP2P, req: PublishRequest
576+
): Future[Result[PublishResponse, string]] {.ffi.} =
577+
let gossipSub = lib.gossipSub.valueOr:
578+
return err("gossipsub not initialized")
579+
let peerCount = await gossipSub.publish(req.topic, req.data)
580+
ok(PublishResponse(peerCount: peerCount))
581+
582+
proc libp2pGossipsubSubscribe*(
583+
lib: LibP2P, topic: string
584+
): Future[Result[bool, string]] {.ffi.} =
585+
let gossipSub = lib.gossipSub.valueOr:
586+
return err("gossipsub not initialized")
587+
if not lib.topicHandlers.hasKey(topic):
588+
let handler = proc(t: string, data: seq[byte]): Future[void] {.async.} =
589+
onPubsubMessage(PubsubMessageEvent(topic: t, data: data))
590+
lib.topicHandlers[topic] = handler
591+
gossipSub.subscribe(topic, handler)
592+
ok(true)
593+
594+
proc libp2pGossipsubUnsubscribe*(
595+
lib: LibP2P, topic: string
596+
): Future[Result[bool, string]] {.ffi.} =
597+
let gossipSub = lib.gossipSub.valueOr:
598+
return err("gossipsub not initialized")
599+
let handler = lib.topicHandlers.getOrDefault(topic, nil)
600+
if not handler.isNil():
601+
lib.topicHandlers.del(topic)
602+
gossipSub.unsubscribe(topic, handler)
603+
ok(true)
604+
561605
genBindings()

0 commit comments

Comments
 (0)