rpc

Documentation

Index

• type Client
  • func NewClient(cluster Cluster) *Client
  • func (c *Client) Call(peerID, action string, payload map[string]interface{}) ([]byte, error)
  • func (c *Client) CallWithContext(ctx context.Context, peerID, action string, payload map[string]interface{}) ([]byte, error)
  • func (c *Client) Close() error
  • func (c *Client) SetAuthToken(token string)
• type Cluster
• type LocalNetworkInterface
• type Logger
• type Node
• type PeerChatHandler
  • func NewPeerChatHandler(cluster Cluster, rpcChannel *channels.RPCChannel) *PeerChatHandler
  • func (h *PeerChatHandler) Handle(payload map[string]interface{}) (map[string]interface{}, error)
• type PeerChatPayload
• type PeerChatResponse
• type RPCHandler
• type RateLimiter
  • func NewRateLimiter(maxTokens int, refillRate time.Duration, maxRequests int, window time.Duration) *RateLimiter
  • func (rl *RateLimiter) Acquire(ctx context.Context, peerID string) error
  • func (rl *RateLimiter) Release(peerID string)
• type Server
  • func NewServer(cluster Cluster) *Server
  • func (s *Server) GetConnectionCount() int
  • func (s *Server) GetPort() int
  • func (s *Server) IsRunning() bool
  • func (s *Server) RegisterHandler(action string, handler RPCHandler)
  • func (s *Server) SetAuthToken(token string)
  • func (s *Server) Start(port int) error
  • func (s *Server) Stop() error
• type TaskResultStorer

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client handles RPC calls to other bots

func NewClient

func NewClient(cluster Cluster) *Client

NewClient creates a new RPC client

func (*Client) Call

func (c *Client) Call(peerID, action string, payload map[string]interface{}) ([]byte, error)

Call makes an RPC call to a peer (deprecated - use CallWithContext for better timeout control)

func (*Client) CallWithContext

func (c *Client) CallWithContext(ctx context.Context, peerID, action string, payload map[string]interface{}) ([]byte, error)

CallWithContext makes an RPC call to a peer with context support

func (*Client) Close

func (c *Client) Close() error

Close closes the client and all connections

func (*Client) SetAuthToken

func (c *Client) SetAuthToken(token string)

SetAuthToken sets the authentication token for RPC connections

type Cluster

type Cluster interface {
	GetRegistry() interface{}
	GetNodeID() string
	GetAddress() string
	GetCapabilities() []string
	GetOnlinePeers() []interface{}
	GetActionsSchema() []interface{} // Get all available actions with schema
	LogRPCInfo(msg string, args ...interface{})
	LogRPCError(msg string, args ...interface{})
	LogRPCDebug(msg string, args ...interface{})
	GetPeer(peerID string) (interface{}, error)                                                                 // Get peer directly
	GetLocalNetworkInterfaces() ([]LocalNetworkInterface, error)                                                // Get local network interfaces
	CallWithContext(ctx context.Context, peerID, action string, payload map[string]interface{}) ([]byte, error) // RPC call
	GetTaskResultStorer() TaskResultStorer                                                                      // H4: B-side task result persistence
}

Cluster represents the cluster interface needed by RPC

type LocalNetworkInterface

type LocalNetworkInterface struct {
	IP   string
	Mask string
}

LocalNetworkInterface represents a local network interface (for RPC interface)

type Logger

type Logger interface {
	RPCInfo(format string, args ...interface{})
	RPCError(format string, args ...interface{})
	RPCDebug(format string, args ...interface{})
}

Logger represents logging functions

type Node

type Node interface {
	GetID() string
	GetName() string
	GetAddress() string
	GetAddresses() []string
	GetRPCPort() int
	GetCapabilities() []string
	GetStatus() string
	IsOnline() bool
}

Node represents a node in the cluster (minimal interface)

type PeerChatHandler

type PeerChatHandler struct {
	// contains filtered or unexported fields
}

PeerChatHandler handles peer-to-peer chat and collaboration requests

func NewPeerChatHandler

func NewPeerChatHandler(cluster Cluster, rpcChannel *channels.RPCChannel) *PeerChatHandler

NewPeerChatHandler creates a new peer chat handler

func (*PeerChatHandler) Handle

func (h *PeerChatHandler) Handle(payload map[string]interface{}) (map[string]interface{}, error)

Handle handles a peer chat request — 立即返回 ACK，异步处理 LLM

type PeerChatPayload

type PeerChatPayload struct {
	Type    string                 `json:"type"`    // chat|request|task|query
	Content string                 `json:"content"` // 对话内容或任务描述
	Context map[string]interface{} `json:"context"` // 附加上下文信息
}

PeerChatPayload represents the payload for peer_chat action

type PeerChatResponse

type PeerChatResponse struct {
	Response string                 `json:"response"`         // 节点的响应内容
	Result   map[string]interface{} `json:"result,omitempty"` // 结构化结果
	Status   string                 `json:"status"`           // success|error|busy
}

PeerChatResponse represents the response from peer_chat action

type RPCHandler

type RPCHandler func(payload map[string]interface{}) (map[string]interface{}, error)

RPCHandler is a function that handles an RPC action

type RateLimiter

type RateLimiter struct {
	// contains filtered or unexported fields
}

RateLimiter limits the rate of RPC calls

func NewRateLimiter

func NewRateLimiter(maxTokens int, refillRate time.Duration, maxRequests int, window time.Duration) *RateLimiter

NewRateLimiter creates a new rate limiter

func (*RateLimiter) Acquire

func (rl *RateLimiter) Acquire(ctx context.Context, peerID string) error

Acquire acquires a token for RPC call to peerID

func (*RateLimiter) Release

func (rl *RateLimiter) Release(peerID string)

Release releases a token after RPC call completes

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server handles incoming RPC requests

func NewServer

func NewServer(cluster Cluster) *Server

NewServer creates a new RPC server

func (*Server) GetConnectionCount

func (s *Server) GetConnectionCount() int

GetConnectionCount returns the number of active connections

func (*Server) GetPort

func (s *Server) GetPort() int

GetPort returns the actual port the server is listening on

func (*Server) IsRunning

func (s *Server) IsRunning() bool

IsRunning returns true if the server is running

func (*Server) RegisterHandler

func (s *Server) RegisterHandler(action string, handler RPCHandler)

RegisterHandler registers an RPC handler for an action

func (*Server) SetAuthToken

func (s *Server) SetAuthToken(token string)

SetAuthToken sets the authentication token for RPC connections

func (*Server) Start

func (s *Server) Start(port int) error

Start starts the RPC server on the given port

func (*Server) Stop

func (s *Server) Stop() error

Stop stops the RPC server

type TaskResultStorer

type TaskResultStorer interface {
	SetRunning(taskID, sourceNode string)
	SetResult(taskID, resultStatus, response, errMsg, sourceNode string) error
	Delete(taskID string) error
}

TaskResultStorer B 端任务结果持久化接口（避免循环依赖，在 rpc 包中定义）