masa

Documentation

Index

• func SubscribeToTopics(node *OracleNode) error
• type NodeDataPage
• type OracleNode
  • func NewOracleNode(ctx context.Context, isStaked bool) (*OracleNode, error)
  • func (node *OracleNode) GetMultiAddrs() multiaddr.Multiaddr
  • func (node *OracleNode) GossipNodeData(stream network.Stream)
  • func (node *OracleNode) HandleMessage(msg *pubsub.Message)
  • func (node *OracleNode) IsPublisher() bool
  • func (node *OracleNode) ListenToNodeTracker()
  • func (node *OracleNode) LogActiveTopics()
  • func (node *OracleNode) ReceiveNodeData(stream network.Stream)
  • func (node *OracleNode) SendNodeData(peerID peer.ID)
  • func (node *OracleNode) SendNodeDataPage(allNodeData []pubsub2.NodeData, stream network.Stream, pageNumber int)
  • func (node *OracleNode) Start() (err error)
  • func (node *OracleNode) Version() string

Functions

func SubscribeToTopics

func SubscribeToTopics(node *OracleNode) error

SubscribeToTopics handles the subscription to various topics for an OracleNode. It subscribes the node to the NodeGossipTopic, AdTopic, and PublicKeyTopic. Each subscription is managed through the node's PubSubManager, which orchestrates the message passing for these topics. Errors during subscription are logged and returned, halting the process to ensure the node's correct setup before operation.

Types

type NodeDataPage

type NodeDataPage struct {
	Data         []pubsub2.NodeData `json:"data"`
	PageNumber   int                `json:"pageNumber"`
	TotalPages   int                `json:"totalPages"`
	TotalRecords int                `json:"totalRecords"`
}

type OracleNode

type OracleNode struct {
	Host     host.Host
	PrivKey  *ecdsa.PrivateKey
	Protocol protocol.ID

	DHT                   *dht.IpfsDHT
	Context               context.Context
	PeerChan              chan myNetwork.PeerEvent
	NodeTracker           *pubsub2.NodeEventTracker
	PubSubManager         *pubsub2.Manager
	Signature             string
	IsStaked              bool
	IsWriter              bool
	IsTwitterScraper      bool
	IsDiscordScraper      bool
	IsWebScraper          bool
	IsLlmServer           bool
	StartTime             time.Time
	AdSubscriptionHandler *ad.SubscriptionHandler
	WorkerTracker         *pubsub2.WorkerEventTracker
	ActorEngine           *actor.RootContext
	ActorRemote           *remote.Remote
	// contains filtered or unexported fields
}

func NewOracleNode

func NewOracleNode(ctx context.Context, isStaked bool) (*OracleNode, error)

NewOracleNode creates a new OracleNode instance with the provided context and staking status. It initializes the libp2p host, DHT, pubsub manager, and other components needed for an Oracle node to join the network and participate.

func (*OracleNode) GetMultiAddrs

func (node *OracleNode) GetMultiAddrs() multiaddr.Multiaddr

GetMultiAddrs returns the priority multiaddr for this node. It first checks if the priority address is already set, and returns it if so. If not, it determines the priority address from the available multiaddrs using the GetPriorityAddress utility function, sets it, and returns it.

func (*OracleNode) GossipNodeData

func (node *OracleNode) GossipNodeData(stream network.Stream)

GossipNodeData handles receiving NodeData from a peer over a network stream. It reads the stream to get the remote peer ID and NodeData, updates the local NodeTracker with the data if it is about another node, and closes the stream when finished.

func (*OracleNode) HandleMessage

func (node *OracleNode) HandleMessage(msg *pubsub.Message)

HandleMessage unmarshals the node data from the pubsub message, and passes it to the NodeTracker to handle. This allows the OracleNode to receive node data published on the network, and process it.

func (*OracleNode) IsPublisher

func (node *OracleNode) IsPublisher() bool

IsPublisher returns true if this node is a publisher node. A publisher node is one that has a non-empty signature.

func (*OracleNode) ListenToNodeTracker

func (node *OracleNode) ListenToNodeTracker()

ListenToNodeTracker listens to the NodeTracker's NodeDataChan and publishes any received node data to the node gossip topic. It also sends the node data directly to the peer if it's a join event and this node isn't a bootnode or has been running for more than 5 minutes.

func (*OracleNode) LogActiveTopics

func (node *OracleNode) LogActiveTopics()

LogActiveTopics logs the currently active topic names to the default logger. It gets the list of active topics from the PubSubManager and logs them if there are any, otherwise it logs that there are no active topics.

func (*OracleNode) ReceiveNodeData

func (node *OracleNode) ReceiveNodeData(stream network.Stream)

ReceiveNodeData handles receiving NodeData pages from a peer over a network stream. It scans the stream and unmarshals each page of NodeData, refreshing the local NodeTracker with the data.

func (*OracleNode) SendNodeData

func (node *OracleNode) SendNodeData(peerID peer.ID)

SendNodeData sends all node data to the peer with the given ID. It first checks if the node is staked, and if not, aborts. It gets the node data to send based on the last time the peer was seen. It paginates the node data into pages and sends each page over the stream.

func (*OracleNode) SendNodeDataPage

func (node *OracleNode) SendNodeDataPage(allNodeData []pubsub2.NodeData, stream network.Stream, pageNumber int)

SendNodeDataPage sends a page of node data over the given stream. It paginates the provided node data slice into pages of size config.PageSize. The pageNumber parameter specifies which page to send, starting from 0. The response includes the page of data, page number, total pages, and total records.

func (*OracleNode) Start

func (node *OracleNode) Start() (err error)

Start initializes the OracleNode by setting up libp2p stream handlers, connecting to the DHT and bootnodes, and subscribing to topics. It launches goroutines to handle discovered peers, listen to the node tracker, and discover peers. If this is a bootnode, it adds itself to the node tracker.

func (*OracleNode) Version

func (node *OracleNode) Version() string

Version returns the current version string of the oracle node software.