masa

Documentation

Index

• func SubscribeToBlocks(ctx context.Context, node *OracleNode)
• func SubscribeToTopics(node *OracleNode) error
• type NodeDataPage
• type OracleNode
  • func NewOracleNode(ctx context.Context, isStaked bool) (*OracleNode, error)
  • func (node *OracleNode) FromUnixTime(unixTime int64) string
  • 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) IsWorker() 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) ToUnixTime(stringTime string) int64
  • func (node *OracleNode) Version() string

Functions

func SubscribeToBlocks

func SubscribeToBlocks(ctx context.Context, node *OracleNode)

SubscribeToBlocks is a function that takes in a context and an OracleNode as parameters. It is used to subscribe the given OracleNode to the blockchain blocks.

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
	IsValidator      bool
	IsTwitterScraper bool
	IsDiscordScraper bool
	IsWebScraper     bool
	IsLlmServer      bool
	StartTime        time.Time
	WorkerTracker    *pubsub2.WorkerEventTracker
	BlockTracker     *pubsub2.BlockEventTracker
	ActorEngine      *actor.RootContext
	ActorRemote      *remote.Remote
	Blockchain       *chain.Chain
	// 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) FromUnixTime

func (node *OracleNode) FromUnixTime(unixTime int64) string

FromUnixTime converts a Unix timestamp into a formatted string. The Unix timestamp is expected to be in seconds. The returned string is in the format "2006-01-02T15:04:05.000Z".

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) IsWorker

func (node *OracleNode) IsWorker() bool

IsWorker determines if the OracleNode is configured to act as an actor. An actor node is one that has at least one of the following scrapers enabled: TwitterScraper, DiscordScraper, or WebScraper. It returns true if any of these scrapers are enabled, otherwise false.

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) ToUnixTime

func (node *OracleNode) ToUnixTime(stringTime string) int64

ToUnixTime converts a formatted string time into a Unix timestamp. The input string is expected to be in the format "2006-01-02T15:04:05.000Z". The returned Unix timestamp is in seconds.

func (*OracleNode) Version

func (node *OracleNode) Version() string

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