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() 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 OracleData
• type Record
• type Worker
  • func (a *Worker) Receive(ctx actor.Context)
• type WorkerType

Constants

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

Variables

var WORKER = struct {
	LLMChat, Twitter, TwitterFollowers, TwitterProfile, TwitterSentiment, TwitterTrends, Web, WebSentiment WorkerType
}{
	LLMChat:          LLMChat,
	Twitter:          Twitter,
	TwitterFollowers: TwitterFollowers,
	TwitterProfile:   TwitterProfile,
	TwitterSentiment: TwitterSentiment,
	TwitterTrends:    TwitterTrends,
	Web:              Web,
	WebSentiment:     WebSentiment,
}

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

func SubscribeToWorkers

func SubscribeToWorkers(node *masa.OracleNode)

SubscribeToWorkers subscribes the given OracleNode to worker events. @note need to add this even if not participating to send messages ....

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 {
	RecordId  string    `json:"cid"`
	Timestamp time.Time `json:"timestamp"`
}

type ChanResponse

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

type OracleData

type OracleData struct {
	Id        string `json:"id"`
	PeerId    string `json:"peer_id"`
	Request   string `json:"request"`
	ModelName string `json:"model_name,omitempty"`
	Steps     []struct {
		Idx               int    `json:"idx"`
		RawContent        string `json:"raw_content,omitempty"`
		StructuredContent string `json:"structured_content,omitempty"`
		SystemPrompt      string `json:"system_prompt,omitempty"`
		Timestamp         string `json:"timestamp"`
		UserPrompt        string `json:"user_prompt,omitempty"`
	} `json:"steps"`
}

type Record

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

type Worker

type Worker struct{}

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 (
	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"
)