workers

Documentation

Index

• Constants
• Variables
• func Get(url string, headers map[string]string) ([]byte, error)
• func MonitorWorkers(ctx context.Context, node *masa.OracleNode)
• func NewWorker(node *masa.OracleNode) actor.Producer
• func Patch(url string, rawJSON json.RawMessage, headers map[string]string) ([]byte, error)
• func Post(url string, rawJSON json.RawMessage, headers map[string]string) ([]byte, error)
• func Put(url string, rawJSON json.RawMessage, headers map[string]string) ([]byte, error)
• func SendWork(node *masa.OracleNode, m *pubsub2.Message)
• func SubscribeToWorkers(node *masa.OracleNode)
• type CID
• type ChanResponse
• type LLMChatBody
• type Record
• type Worker
  • func (a *Worker) HandleConnect(ctx actor.Context, m *messages.Connect)
  • func (a *Worker) HandleLog(ctx actor.Context, l string)
  • func (a *Worker) HandleWork(ctx actor.Context, m *messages.Work, node *masa.OracleNode)
  • func (a *Worker) Receive(ctx actor.Context)
• type WorkerType

Constants

const (
	GET   = "GET"
	POST  = "POST"
	PUT   = "PUT"
	PATCH = "PATCH"
)

Variables

var WORKER = struct {
	Discord, DiscordProfile, DiscordChannelMessages, DiscordGuildChannels, DiscordUserGuilds, LLMChat, Twitter, TwitterFollowers, TwitterProfile, TwitterSentiment, TwitterTrends, Web, WebSentiment, Test WorkerType
}{
	Discord:                Discord,
	DiscordProfile:         DiscordProfile,
	DiscordChannelMessages: DiscordChannelMessages,
	DiscordGuildChannels:   DiscordGuildChannels,
	DiscordUserGuilds:      DiscordUserGuilds,
	LLMChat:                LLMChat,
	Twitter:                Twitter,
	TwitterFollowers:       TwitterFollowers,
	TwitterProfile:         TwitterProfile,
	TwitterSentiment:       TwitterSentiment,
	TwitterTrends:          TwitterTrends,
	Web:                    Web,
	WebSentiment:           WebSentiment,
	Test:                   Test,
}

Functions

func Get

func Get(url string, headers map[string]string) ([]byte, error)

func MonitorWorkers

func MonitorWorkers(ctx context.Context, node *masa.OracleNode)

MonitorWorkers monitors worker data by subscribing to the completed work topic, computing a CID for each received data, and writing the data to the database.

Parameters:

• ctx: The context for the monitoring operation.
• node: A pointer to the OracleNode instance.

The function uses a ticker to periodically log a debug message every 60 seconds. It subscribes to the completed work topic using the PubSubManager and handles the received data. For each received data, it computes a CID using the computeCid function, logs the CID, marshals the data to JSON, and writes it to the database using the WriteData function. The monitoring continues until the context is done.

func NewWorker

func NewWorker(node *masa.OracleNode) actor.Producer

NewWorker creates a new instance of the Worker actor. It implements the actor.Receiver interface, allowing it to receive and handle messages.

Returns:

• An instance of the Worker struct that implements the actor.Receiver interface.

func Patch

func Patch(url string, rawJSON json.RawMessage, headers map[string]string) ([]byte, error)

func Post

func Post(url string, rawJSON json.RawMessage, headers map[string]string) ([]byte, error)

func Put

func Put(url string, rawJSON json.RawMessage, headers map[string]string) ([]byte, error)

func SendWork

func SendWork(node *masa.OracleNode, m *pubsub2.Message)

SendWork is a function that sends work to a node. It takes two parameters: node: A pointer to a masa.OracleNode object. This is the node to which the work will be sent. m: A pointer to a pubsub2.Message object. This is the message that contains the work to be sent.

func SubscribeToWorkers

func SubscribeToWorkers(node *masa.OracleNode)

SubscribeToWorkers subscribes the given OracleNode to worker events.

Parameters:

• node: A pointer to the OracleNode instance that will be subscribed to worker events.

The function initializes the WorkerEventTracker for the node and adds a subscription to the worker topic using the PubSubManager. If an error occurs during the subscription, it logs the error.

Types

type CID

type CID struct {
	Duration  float64   `json:"duration"`
	RecordId  string    `json:"cid"`
	Timestamp time.Time `json:"timestamp"`
}

type ChanResponse

type ChanResponse struct {
	Response  map[string]interface{}
	ChannelId string
}

type LLMChatBody

type LLMChatBody struct {
	Model    string `json:"model,omitempty"`
	Messages []struct {
		Role    string `json:"role"`
		Content string `json:"content"`
	} `json:"messages,omitempty"`
	Stream bool `json:"stream"`
}

type Record

type Record struct {
	PeerId string `json:"peerid"`
	CIDs   []CID  `json:"cids"`
}

type Worker

type Worker struct {
	Node *masa.OracleNode
}

func (*Worker) HandleConnect

func (a *Worker) HandleConnect(ctx actor.Context, m *messages.Connect)

HandleConnect is a method of the Worker struct that handles the connection of a worker. It takes in an actor context and a Connect message as parameters.

func (*Worker) HandleLog

func (a *Worker) HandleLog(ctx actor.Context, l string)

HandleLog is a method of the Worker struct that handles logging. It takes in an actor context and a string message as parameters.

func (*Worker) HandleWork

func (a *Worker) HandleWork(ctx actor.Context, m *messages.Work, node *masa.OracleNode)

HandleWork is a method of the Worker struct that handles the work assigned to a worker. It takes in an actor context and a Work message as parameters.

func (*Worker) Receive

func (a *Worker) Receive(ctx actor.Context)

Receive is the message handling method for the Worker actor. It receives messages through the actor context and processes them based on their type.

type WorkerType

type WorkerType string

const (
	Discord                WorkerType = "discord"
	DiscordProfile         WorkerType = "discord-profile"
	DiscordChannelMessages WorkerType = "discord-channel-messages"
	DiscordGuildChannels   WorkerType = "discord-guild-channels"
	DiscordUserGuilds      WorkerType = "discord-user-guilds"
	LLMChat                WorkerType = "llm-chat"
	Twitter                WorkerType = "twitter"
	TwitterFollowers       WorkerType = "twitter-followers"
	TwitterProfile         WorkerType = "twitter-profile"
	TwitterSentiment       WorkerType = "twitter-sentiment"
	TwitterTrends          WorkerType = "twitter-trends"
	Web                    WorkerType = "web"
	WebSentiment           WorkerType = "web-sentiment"
	Test                   WorkerType = "test"
)