/// # Tutorial 4: Custom Protocol Handlers /// /// In the [previous tutorial](tutorial_3_connecting_peers.md), we used the /// built-in Ping protocol to exchange data between peers. But real /// applications need their own protocols! /// /// This tutorial shows you how to mount a custom protocol on a node so /// that it can handle incoming streams from peers who dial that protocol. /// /// ## How Custom Protocols Work /// /// A protocol in libp2p is identified by a **protocol ID string** — a /// `/`-separated path like `/myapp/chat/1.0.0`. When a remote peer dials /// this protocol ID, your node receives a new stream. /// /// To handle incoming streams you: /// 1. Call `mountProtocol()` to register a protocol ID /// 2. Set an `emitEvent` callback that listens for `"protocolStream"` events /// 3. Read from and write to the stream in the event handler ⚠️ /// (see performance notes below) /// /// The stream lifecycle on the server side is: /// 1. Receive `protocolStream` event with a `streamId` /// 2. Read data from the stream /// 3. Write data to the stream (optional) /// 4. Call `streamRelease()` when done with the stream /// /// > **Note**: Unlike the dialing side, the server side does **not** call /// > `streamClose()` — the peer that initiated the stream is responsible /// > for closing it. /// /// ----------- #include #include #include #include #include #include #include #include "plugin.h" using json = nlohmann::json; /// ## Defining our custom protocol /// /// We'll create an **echo protocol**: the server reads a length-prefixed /// message and echoes it back to the client. This is a common pattern /// for request-response protocols. const std::string kEchoProtocol = "/examples/echo/1.0.0"; int main() { printf("=== Tutorial 4: Custom Protocol Handlers ===\n\n"); setLogLevel("fatal"); /// ## Step 1: Create and start two nodes Libp2pModuleOptions optsA, optsB; optsA.addrs = {"/ip4/127.0.0.1/tcp/9290"}; optsB.addrs = {"/ip4/127.0.0.1/tcp/9291"}; Libp2pModuleImpl nodeA(optsA); Libp2pModuleImpl nodeB(optsB); StdLogosResult startARes = nodeA.start(); if (!startARes.success) { fprintf(stderr, "Node A failed: %s\n", startARes.error.c_str()); return 1; } StdLogosResult startBRes = nodeB.start(); if (!startBRes.success) { fprintf(stderr, "Node B failed: %s\n", startBRes.error.c_str()); return 1; } printf("Both nodes started\n"); /// ## Step 2: Set up the protocol handler on Node A /// /// We define an `emitEvent` callback on Node A. Whenever a remote peer /// dials our protocol, a `"protocolStream"` event fires with a JSON /// payload containing the `streamId`. /// /// For simplicity, this example reads one message, echoes it, then /// releases the stream. A real application would likely spawn a /// background thread to handle concurrent streams. struct ServerState { std::mutex mtx; std::condition_variable cv; uint64_t streamId = 0; bool ready = false; }; ServerState server; nodeA.emitEvent = [&](const std::string& name, const std::string& data) { if (name != "protocolStream") return; auto j = json::parse(data); uint64_t sid = j["streamId"].get(); { std::lock_guard lock(server.mtx); server.streamId = sid; server.ready = true; } server.cv.notify_one(); }; /// Register the echo protocol on Node A: printf("Mounting protocol '%s' on Node A...\n", kEchoProtocol.c_str()); StdLogosResult mountRes = nodeA.mountProtocol(kEchoProtocol); if (!mountRes.success) { fprintf(stderr, "Failed to mount protocol: %s\n", mountRes.error.c_str()); return 1; } /// ## Step 3: Get Node A's address and connect Node B StdLogosResult infoARes = nodeA.peerInfo(); if (!infoARes.success) { fprintf(stderr, "Failed to get Node A info: %s\n", infoARes.error.c_str()); return 1; } auto infoA = infoARes.value; std::string peerIdA = infoA["peerId"].get(); std::vector addrsA; for (const auto& a : infoA["addrs"]) addrsA.push_back(a.get()); printf("Connecting Node B to Node A...\n"); StdLogosResult connectRes = nodeB.connectPeer(peerIdA, addrsA, 5000); if (!connectRes.success) { fprintf(stderr, "Failed to connect: %s\n", connectRes.error.c_str()); return 1; } printf("Connected\n"); /// ## Step 4: Node B dials the echo protocol /// /// When Node B dials our custom protocol, Node A's protocol handler /// fires, and Node A receives a new stream. printf("Node B dialing '%s'...\n", kEchoProtocol.c_str()); StdLogosResult dialRes = nodeB.dial(peerIdA, kEchoProtocol); if (!dialRes.success) { fprintf(stderr, "Dial failed: %s\n", dialRes.error.c_str()); return 1; } uint64_t clientStreamId = dialRes.value.get(); printf("Node B client stream id: %llu\n", (unsigned long long)clientStreamId); /// ## Step 5: Wait for Node A to receive the stream uint64_t serverStreamId = 0; { std::unique_lock lock(server.mtx); if (!server.cv.wait_for(lock, std::chrono::seconds(5), [&] { return server.ready; })) { fprintf(stderr, "Timed out waiting for incoming stream\n"); return 1; } serverStreamId = server.streamId; } printf("Node A received stream id: %llu\n", (unsigned long long)serverStreamId); /// ## Step 6: Node B sends a message std::string message = "Hello from Node B!"; printf("Node B sending: \"%s\"\n", message.c_str()); StdLogosResult writeRes = nodeB.streamWriteLp(clientStreamId, message); if (!writeRes.success) { fprintf(stderr, "Write failed: %s\n", writeRes.error.c_str()); return 1; } /// ## Step 7: Node A reads the message and echoes it back StdLogosResult readRes = nodeA.streamReadLp(serverStreamId, 4096); if (!readRes.success) { fprintf(stderr, "Node A read failed: %s\n", readRes.error.c_str()); return 1; } std::string received = base64Decode(readRes.value.get()); printf("Node A received: \"%s\"\n", received.c_str()); /// Echo it back: StdLogosResult echoWriteRes = nodeA.streamWriteLp(serverStreamId, received); if (!echoWriteRes.success) { fprintf(stderr, "Node A echo write failed: %s\n", echoWriteRes.error.c_str()); return 1; } /// ## Step 8: Node B reads the echo StdLogosResult echoRes = nodeB.streamReadLp(clientStreamId, 4096); if (!echoRes.success) { fprintf(stderr, "Node B read echo failed: %s\n", echoRes.error.c_str()); return 1; } std::string echo = base64Decode(echoRes.value.get()); printf("Node B received echo: \"%s\"\n", echo.c_str()); if (echo != message) { fprintf(stderr, "Echo mismatch! Got '%s'\n", echo.c_str()); return 1; } printf("Echo verified successfully!\n"); /// ## Step 9: Clean up /// /// The initiator (the peer that dialed) is responsible for closing the /// stream. The responder only needs to release their handle: // Server side (responder) - just release the handle nodeA.streamRelease(serverStreamId); // Client side (initiator) - close with EOF, then release nodeB.streamCloseWithEOF(clientStreamId); nodeB.streamRelease(clientStreamId); nodeA.stop(); nodeB.stop(); printf("\n=== Tutorial 4 Complete ===\n"); return 0; } /// ## Key Takeaways /// /// - `mountProtocol()` registers a handler on the server side /// - `emitEvent` with the `"protocolStream"` event delivers incoming streams /// - The dialing side uses `streamClose()`/`streamCloseWithEOF()` /// - The server side uses `streamRelease()` (no close) /// - Use `streamWriteLp()` / `streamReadLp()` for length-prefixed messages /// ## Important: Performance Considerations /// /// For production code, avoid blocking reads/writes in event handlers. /// Instead, use asynchronous patterns where: /// 1. Event handler quickly passes stream to a queue/worker /// 2. Separate mechanism handles the actual I/O /// 3. Event loop stays responsive for other connections /// /// The tutorial is simplified for learning; real implementations /// should use non-blocking patterns to maintain system responsiveness. /// ## Run tutorial /// /// ```bash /// ./build/tutorial/tutorial_4_custom_protocol /// ```