subpub

README

SubPub

SubPub is a simplified PubSub protocol using libp2p

• Creates a libp2p Host, with a PubSub service attached that uses the GossipSub router.
• method SendBroadcast() uses this PubSub service for routing broadcast messages to peers.
• method SendUnicast() sends a unicast packet directly to a single peer.
• The topic used for p2p discovery is determined from the groupKey (hash)
• Incoming messages (both unicasts and broadcasts) are passed to the messages channel indicated in Start()

Documentation

Index

• Constants
• type Gossip
  • func (g *Gossip) Publish(message []byte) error
• type Message
• type SubPub
  • func NewSubPub(groupKey [32]byte, node *core.IpfsNode) *SubPub
  • func (s *SubPub) AddPeer(peer string) error
  • func (s *SubPub) Address() string
  • func (s *SubPub) Close()
  • func (s *SubPub) SendBroadcast(data []byte) error
  • func (s *SubPub) SendUnicast(address string, data []byte) error
  • func (s *SubPub) Start(ctx context.Context, receiver chan *Message)
  • func (s *SubPub) Stats() map[string]any
  • func (s *SubPub) String() string

Constants

const (
	// UnicastBufSize is the number of unicast incoming messages to buffer.
	UnicastBufSize = 128

	// DiscoveryPeriodSeconds is the time between discovery rounds.
	DiscoveryPeriodSeconds = 120
)

const GossipBufSize = 128

GossipBufSize is the number of incoming messages to buffer for each topic.

Types

type Gossip

type Gossip struct {
	// Messages is a channel of messages received from other peers in the topic
	Messages chan *Message
	// contains filtered or unexported fields
}

Gossip represents a subscription to a single PubSub topic. Messages can be published to the topic with Gossip.Publish, and received messages are pushed to the Messages channel.

func (*Gossip) Publish

func (g *Gossip) Publish(message []byte) error

Publish sends a message to the gossip pubsub topic.

type Message

type Message struct {
	Data []byte
	Peer string
}

Message is the type of messages passed around by SubPub.

type SubPub

type SubPub struct {
	Key         ecdsa.PrivateKey
	GroupKey    [32]byte
	Topic       string
	NoBootStrap bool
	BootNodes   []string
	NodeID      string

	MaxDHTpeers  int
	OnlyDiscover bool

	DiscoveryPeriod time.Duration

	// These are used in testing
	OnPeerAdd    func(id libpeer.ID)
	OnPeerRemove func(id libpeer.ID)
	// contains filtered or unexported fields
}

SubPub is a simplified PubSub protocol using libp2p. It uses a shared secret to encrypt messages. The shared secret is derived from a group key, which is also used to discover peers. It allows for broadcasting messages to all peers in the group, and sending messages to specific peers. Peers are discovered using the DHT, and automatically set as protected peers.

func NewSubPub

func NewSubPub(groupKey [32]byte, node *core.IpfsNode) *SubPub

NewSubPub creates a new SubPub instance. The groupKey is a secret shared among the PubSub participants. Only those with the key will be able to join.

func (*SubPub) AddPeer

func (s *SubPub) AddPeer(peer string) error

AddPeer creates a new libp2p peer connection (transport layer)

func (*SubPub) Address

func (s *SubPub) Address() string

Address returns the node's ID.

func (*SubPub) Close

func (s *SubPub) Close()

Close terminates the subpub networking stack

func (*SubPub) SendBroadcast

func (s *SubPub) SendBroadcast(data []byte) error

SendBroadcast sends a message to all peers in the cluster.

func (*SubPub) SendUnicast

func (s *SubPub) SendUnicast(address string, data []byte) error

SendUnicast sends a message to a specific peer in the cluster.

func (*SubPub) Start

func (s *SubPub) Start(ctx context.Context, receiver chan *Message)

Start connects the SubPub networking stack and begins passing incoming messages to the receiver chan

func (*SubPub) Stats

func (s *SubPub) Stats() map[string]any

Stats returns the current stats of the SubPub instance.

func (*SubPub) String

func (s *SubPub) String() string

String returns a string representation of the SubPub instance.