sgns::crdt::PubSubBroadcasterExt

Extended PubSub broadcaster that integrates with a CRDT datastore and Graphsync DAG syncer. More...

#include <pubsub_broadcaster_ext.hpp>

Inherits from sgns::crdt::Broadcaster, std::enable_shared_from_this< PubSubBroadcasterExt >

Public Types

	Name
using sgns::ipfs_pubsub::GossipPubSub	GossipPubSub

Public Functions

	Name
	~PubSubBroadcasterExt()
virtual outcome::result< void >	Broadcast(const base::Buffer & buff, std::string topic, boost::optional< libp2p::peer::PeerInfo > peerInfo =boost::none) override Sends the given buffer as a broadcast to peers.
virtual outcome::result< base::Buffer >	Next() override Retrieves the next incoming broadcast payload.
virtual void	WaitForNext(std::chrono::milliseconds timeout) override Blocks until a message is queued or timeout elapses.
virtual void	CancelWait() override Releases any waiter and makes later waits return at once.
void	Start() Subscribes to all configured topics and starts message processing. Must be called before using Next() to receive incoming messages.
outcome::result< void >	AddBroadcastTopic(const std::string & topicName) Adds a new topic by name.
void	AddListenTopic(std::string topic) Subscribe to a given topic and store its future.
virtual bool	HasTopic(const std::string & topic) override Checks whether the given topic is already registered.
virtual std::shared_ptr< void >	GetDagSyncer() const override Get the underlying GraphsyncDAGSyncer instance.
void	Stop()
bool	AddSingleCIDInfo(const std::string & cid, const std::string peer_id, const std::string address)
std::shared_ptr< PubSubBroadcasterExt >	New(std::shared_ptr< sgns::crdt::GraphsyncDAGSyncer > dagSyncer, std::shared_ptr< GossipPubSub > pubSub) Factory method to create a broadcaster for multiple topics.

Additional inherited members

Public Types inherited from sgns::crdt::Broadcaster

	Name
enum class	ErrorCode

Public Functions inherited from sgns::crdt::Broadcaster

	Name
virtual	~Broadcaster() =default

Detailed Description

class sgns::crdt::PubSubBroadcasterExt;

Extended PubSub broadcaster that integrates with a CRDT datastore and Graphsync DAG syncer.

Manages multiple gossip topics, broadcasting messages and processing incoming payloads.

Public Types Documentation

using GossipPubSub

using sgns::crdt::PubSubBroadcasterExt::GossipPubSub =  sgns::ipfs_pubsub::GossipPubSub;

Public Functions Documentation

function ~PubSubBroadcasterExt

~PubSubBroadcasterExt()

function Broadcast

virtual outcome::result< void > Broadcast(
    const base::Buffer & buff,
    std::string topic,
    boost::optional< libp2p::peer::PeerInfo > peerInfo =boost::none
) override

Sends the given buffer as a broadcast to peers.

Parameters:

• buff Buffer containing the data to broadcast.
• topic Topic to broadcast to.
• peerInfo Optional peer info to avoid repeated GetPeerInfo calls.

Return: outcome::success on successful publish, or outcome::failure on error.

Reimplements: sgns::crdt::Broadcaster::Broadcast

function Next

virtual outcome::result< base::Buffer > Next() override

Retrieves the next incoming broadcast payload.

Return: buffer value or outcome::failure on error

Reimplements: sgns::crdt::Broadcaster::Next

function WaitForNext

virtual void WaitForNext(
    std::chrono::milliseconds timeout
) override

Blocks until a message is queued or timeout elapses.

Parameters:

• timeout Longest time to block.

Reimplements: sgns::crdt::Broadcaster::WaitForNext

function CancelWait

virtual void CancelWait() override

Releases any waiter and makes later waits return at once.

Reimplements: sgns::crdt::Broadcaster::CancelWait

function Start

void Start()

Subscribes to all configured topics and starts message processing. Must be called before using Next() to receive incoming messages.

Note: Ensures message processing is ready before any CRDT operations run.

function AddBroadcastTopic

outcome::result< void > AddBroadcastTopic(
    const std::string & topicName
)

Adds a new topic by name.

Parameters:

• topicName Name of the topic to add.

Return: outcome::success() on success (or if topic already existed), outcome::failure() on error.

function AddListenTopic

void AddListenTopic(
    std::string topic
)

Subscribe to a given topic and store its future.

Parameters:

• topic Name of the topic to listen to.

function HasTopic

virtual bool HasTopic(
    const std::string & topic
) override

Checks whether the given topic is already registered.

Parameters:

• topic Name of the topic to check.

Return: True if the topic exists, false otherwise.

Reimplements: sgns::crdt::Broadcaster::HasTopic

function GetDagSyncer

inline virtual std::shared_ptr< void > GetDagSyncer() const override

Get the underlying GraphsyncDAGSyncer instance.

Return: Shared pointer to the GraphsyncDAGSyncer (as void pointer).

Reimplements: sgns::crdt::Broadcaster::GetDagSyncer

function Stop

void Stop()

function AddSingleCIDInfo

bool AddSingleCIDInfo(
    const std::string & cid,
    const std::string peer_id,
    const std::string address
)

function New

static std::shared_ptr< PubSubBroadcasterExt > New(
    std::shared_ptr< sgns::crdt::GraphsyncDAGSyncer > dagSyncer,
    std::shared_ptr< GossipPubSub > pubSub
)

Factory method to create a broadcaster for multiple topics.

Parameters:

• dagSyncer Graphsync DAG syncer for block exchange.
• pubSub PubSub instance used to subscribe and publish.

Return: Shared pointer to the new PubSubBroadcasterExt.

---

Updated on 2026-08-26 at 21:24:47 +0000