feat: add gossipsub (#27)

Author: gmelodie

CMakeLists.txt

@@ -26,6 +26,7 @@ logos_module(
         src/kademlia.cpp
         src/stream.cpp
         src/mix.cpp
+        src/gossipsub.cpp
     EXTERNAL_LIBS
         libp2p
     INCLUDE_DIRS

README.md

@@ -6,6 +6,8 @@ It provides:
 - Peer connectivity
 - Stream management
 - Kademlia DHT operations
+- Mix operaions
+- Gossipsub operaions
 - Sync and async APIs compatible with Qt
 
 Check the examples/ directory for complete usage demonstrations.

examples/README.md

@@ -23,6 +23,12 @@ use cases using the `logos-libp2p-module`.
   - Populating mix pools
   - Dialing through mix
 
+- [Gossipsub](gossipsub.cpp) — Demonstrates usage of the [Gossipsub](https://github.com/libp2p/specs/blob/master/pubsub/gossipsub/README.md) protocol for messaging:
+  - Connecting to a peer
+  - Subscribing to a topic
+  - Publishing to that topic
+  - Receiving the message on the subscribed topic
+
 ## Building
 Enter the development shell and configure cmake
 
@@ -43,5 +49,6 @@ After building, run the executable from the build directory:
 ./build/examples/example_ping
 ./build/examples/example_kademlia
 ./build/examples/example_mix
+./build/examples/example_gossipsub
 ```
 

examples/gossipsub.cpp

@@ -0,0 +1,77 @@
+#include <QCoreApplication>
+#include <QDebug>
+#include <QThread>
+#include "plugin.h"
+
+int main(int argc, char *argv[])
+{
+    QCoreApplication app(argc, argv);
+
+    /* -----------------------------
+       Node A setup
+       ----------------------------- */
+    Libp2pModulePlugin nodeA;
+    Libp2pModulePlugin nodeB;
+
+    qDebug() << "Starting Nodes...";
+    if (!nodeA.syncLibp2pStart().ok) {
+        qFatal("Node A failed to start");
+    }
+    if (!nodeB.syncLibp2pStart().ok) {
+        qFatal("Node B failed to start");
+    }
+
+    PeerInfo nodeAPeerInfo = nodeA.syncPeerInfo().data.value<PeerInfo>();
+
+    if (!nodeB.syncConnectPeer(nodeAPeerInfo.peerId, nodeAPeerInfo.addrs, 500).ok) {
+        qFatal("Node B failed to connect to Node A");
+    }
+
+    /* -----------------------------
+       Subscribe nodes to topic
+       ----------------------------- */
+    QString topic = "demo-topic";
+    qDebug() << "Node B subscribing to topic:" << topic;
+    if (!nodeB.syncGossipsubSubscribe(topic).ok) {
+        qFatal("Node B subscription failed");
+    }
+    qDebug() << "Node A subscribing to topic:" << topic;
+    if (!nodeA.syncGossipsubSubscribe(topic).ok) {
+        qFatal("Node A subscription failed");
+    }
+
+    /* -----------------------------
+       Give the mesh time to form
+       ----------------------------- */
+    qDebug() << "Waiting for mesh to form";
+    QThread::msleep(2000);
+
+    /* -----------------------------
+       Publish message from Node A
+       ----------------------------- */
+    QByteArray payload = "Hello from Node A via gossipsub!";
+    qDebug() << "Node A publishing message to topic:" << topic;
+    if (!nodeA.syncGossipsubPublish(topic, payload).ok) {
+        qFatal("Node A publish failed");
+    }
+
+    /* -----------------------------
+       Fetch messages for Node B
+       ----------------------------- */
+    auto res = nodeB.syncGossipsubNextMessage(topic);
+    if (!res.ok) {
+        qFatal("Node B did not receive any messages");
+    }
+    QByteArray message = res.data.value<QByteArray>();
+    qDebug() << "Node B received:" << message;
+
+    /* -----------------------------
+       Cleanup
+       ----------------------------- */
+    nodeA.syncLibp2pStop();
+    nodeB.syncLibp2pStop();
+
+    qDebug() << "Done";
+
+    return 0;
+}

flake.lock

@@ -66,7 +66,10 @@ void Libp2pModulePlugin::libp2pCallback(
         Qt::QueuedConnection
     );
 
-    delete callbackCtx;
+    // topicHandler is called with context
+    // cannot free it here if that's the case
+    if (caller != "gossipsubSubscribe")
+        delete callbackCtx;
 }
 
 QList<ServiceInfo> copyServiceInfos(
@@ -471,3 +474,49 @@ void Libp2pModulePlugin::connectionCallback(
     delete callbackCtx;
 }
 
+
+void Libp2pModulePlugin::topicHandler(
+    const char *topic,
+    uint8_t *data,
+    size_t len,
+    void *userData
+)
+{
+    auto *callbackCtx = static_cast<CallbackContext *>(userData);
+    if (!callbackCtx) return;
+
+    Libp2pModulePlugin *self = callbackCtx->instance;
+    if (!self) { delete callbackCtx; return; }
+
+    QString topicStr = QString::fromUtf8(topic);
+    QByteArray payload(reinterpret_cast<const char*>(data), static_cast<int>(len));
+
+    // add payload to the queue for syncGossipsubNextMessage
+    {
+        QMutexLocker lock(&self->m_queueMutex);
+
+        // ensure a queue exists for this topic
+        if (!self->m_topicQueues.contains(topicStr)) {
+            self->m_topicQueues[topicStr] = QSharedPointer<QQueue<QByteArray>>::create();
+        }
+
+        self->m_topicQueues[topicStr]->enqueue(payload);
+        self->m_queueCond.wakeOne();
+    }
+
+    QPointer<Libp2pModulePlugin> safeSelf(self);
+    QMetaObject::invokeMethod(
+        safeSelf,
+        [safeSelf, topicStr, payload]() {
+            if (!safeSelf) return;
+            emit safeSelf->libp2pEvent(
+                RET_OK,
+                QString{},                // no request id (push event)
+                QString("gossipsubMessage"),
+                topicStr,
+                QVariant(payload)
+            );
+        },
+        Qt::QueuedConnection
+    );
+}

src/callbacks.cpp

@@ -0,0 +1,95 @@
+#include "plugin.h"
+
+#include <QtCore/QUuid>
+#include <QtCore/QByteArray>
+#include <QtCore/QDebug>
+#include <cstring>
+
+QString Libp2pModulePlugin::gossipsubPublish(
+    const QString &topic,
+    const QByteArray &data
+)
+{
+    if (!ctx)
+        return {};
+
+    QString uuid = QUuid::createUuid().toString();
+    auto *callbackCtx =
+        new CallbackContext{ "gossipsubPublish", uuid, this };
+
+    QByteArray topicUtf8 = topic.toUtf8();
+
+    int ret = libp2p_gossipsub_publish(
+        ctx,
+        topicUtf8.constData(),
+        reinterpret_cast<uint8_t*>(const_cast<char*>(data.constData())),
+        data.size(),
+        &Libp2pModulePlugin::libp2pCallback,
+        callbackCtx
+    );
+
+    if (ret != RET_OK) {
+        delete callbackCtx;
+        return {};
+    }
+
+    return uuid;
+}
+
+QString Libp2pModulePlugin::gossipsubSubscribe(
+    const QString &topic
+)
+{
+    if (!ctx)
+        return {};
+
+    QString uuid = QUuid::createUuid().toString();
+    auto *callbackCtx =
+        new CallbackContext{ "gossipsubSubscribe", uuid, this };
+
+    QByteArray topicUtf8 = topic.toUtf8();
+
+    int ret = libp2p_gossipsub_subscribe(
+        ctx,
+        topicUtf8.constData(),
+        &Libp2pModulePlugin::topicHandler,
+        &Libp2pModulePlugin::libp2pCallback,
+        callbackCtx
+    );
+
+    if (ret != RET_OK) {
+        delete callbackCtx;
+        return {};
+    }
+
+    return uuid;
+}
+
+QString Libp2pModulePlugin::gossipsubUnsubscribe(
+    const QString &topic
+)
+{
+    if (!ctx)
+        return {};
+
+    QString uuid = QUuid::createUuid().toString();
+    auto *callbackCtx =
+        new CallbackContext{ "gossipsubUnsubscribe", uuid, this };
+
+    QByteArray topicUtf8 = topic.toUtf8();
+
+    int ret = libp2p_gossipsub_unsubscribe(
+        ctx,
+        topicUtf8.constData(),
+        &Libp2pModulePlugin::topicHandler,
+        &Libp2pModulePlugin::libp2pCallback,
+        callbackCtx
+    );
+
+    if (ret != RET_OK) {
+        delete callbackCtx;
+        return {};
+    }
+
+    return uuid;
+}

src/gossipsub.cpp

@@ -145,9 +145,9 @@ class Libp2pModuleInterface : public PluginInterface
     Q_INVOKABLE virtual QString peerInfo() = 0;
 
     /// Returns currently connected peers.
-    /// Possible Direction values: Direction_In = 0, Direction_Out = 1.
+    /// Possible Direction values: Direction_In, Direction_Out.
     /// Returns a UUID string identifying this request.
-    Q_INVOKABLE virtual QString connectedPeers(int direction = 0) = 0;
+    Q_INVOKABLE virtual QString connectedPeers(int direction = Direction_In) = 0;
 
     /// Opens a stream using a protocol.
     /// Returns a UUID string identifying this request.
@@ -214,6 +214,48 @@ class Libp2pModuleInterface : public PluginInterface
     /// data: none
     Q_INVOKABLE virtual Libp2pResult syncStreamRelease(uint64_t streamId) = 0;
 
+    /* ----------- Gossipsub ----------- */
+
+    /// Publish data to a topic
+    /// Returns a UUID string identifying this request.
+    Q_INVOKABLE virtual QString gossipsubPublish(
+        const QString &topic,
+        const QByteArray &data
+    ) = 0;
+
+    /// Subscribe to a topic
+    /// Returns a UUID string identifying this request.
+    Q_INVOKABLE virtual QString gossipsubSubscribe(
+        const QString &topic
+    ) = 0;
+
+    /// Unsubscribe from a topic
+    /// Returns a UUID string identifying this request.
+    Q_INVOKABLE virtual QString gossipsubUnsubscribe(
+        const QString &topic
+    ) = 0;
+
+    /* ----------- Sync Gossipsub ----------- */
+
+    /// data: none
+    Q_INVOKABLE virtual Libp2pResult syncGossipsubPublish(
+        const QString &topic,
+        const QByteArray &data
+    ) = 0;
+    /// data: none
+    Q_INVOKABLE virtual Libp2pResult syncGossipsubSubscribe(
+        const QString &topic
+    ) = 0;
+    /// data: none
+    Q_INVOKABLE virtual Libp2pResult syncGossipsubUnsubscribe(
+        const QString &topic
+    ) = 0;
+    /// data: QByteArray (message)
+    Q_INVOKABLE virtual Libp2pResult syncGossipsubNextMessage(
+        const QString &topic,
+        int timeoutMs
+    ) = 0;
+
     /* ----------- Kademlia ----------- */
 
     /// Converts a key to a CID.