docs(comments): 注释中文化 · internal/runtime/orchestration(29 文件,纯替换)
- 只翻注释/文档字符串:剥离注释后代码行逐字节不变(机械校验只剩允许的行尾注释形态) - 裁决编号与 design 节号(02.4.3 / 11.4.2 / design/05 §9 等)逐字保留 - 复核:go build 绿;orchestration 及其子包(audit/comm)测试绿;gofmt 干净
This commit is contained in:
1 parent
fe90f64a56
commit
eb1828ee7a
29 files changed
+751
-949
No files matched your search
@@ -1,42 +1,36 @@
|
||||
// Package comm implements agent-to-agent communication for the Cyrene v2
|
||||
// orchestration layer (DevDocs/design/05-子代理.md §4). Subagents and the main
|
||||
// agent exchange messages over the in-process bus (internal/bus) using a
|
||||
// request/answer protocol built on three message kinds: ask, reply and event.
|
||||
// Package comm 为 Cyrene v2 编排层实现 agent 间通信
|
||||
// (DevDocs/design/05-子代理.md §4)。子代理与主 agent 经进程内总线(internal/bus)
|
||||
// 交换消息,所用请求/应答协议建立在三种消息之上:ask、reply 与 event。
|
||||
//
|
||||
// # Protocol
|
||||
// # 协议
|
||||
//
|
||||
// A Probe is bound to one agent id. It can ask another agent a question and
|
||||
// wait for its reply (ask/reply), or fire events at another agent or to anyone
|
||||
// listening (event). Messages are published to the bus with dot-separated
|
||||
// subjects so the bus's NATS-like wildcard matching routes them to the right
|
||||
// agent; see the "Subject convention" section below.
|
||||
// 一个 Probe 绑定到一个 agent id。它可以向另一个 agent 提问并等待其回复
|
||||
// (ask/reply),也可以向另一个 agent 或任何监听者发射事件(event)。消息以点分隔的
|
||||
// subject 发布到总线上,因此总线类似 NATS 的通配匹配能把它们路由到正确的 agent;
|
||||
// 见下方「Subject 约定」一节。
|
||||
//
|
||||
// # Subject convention
|
||||
// # Subject 约定
|
||||
//
|
||||
// Subjects are built from sanitized agent ids (any rune outside [A-Za-z0-9_-]
|
||||
// is replaced by "_", so an id never breaks a subject token):
|
||||
// Subject 由经净化的 agent id 拼成(凡不在 [A-Za-z0-9_-] 之内的字符都被替换为 "_",
|
||||
// 因此 id 绝不会破坏 subject 词元):
|
||||
//
|
||||
// ask: cyrene.agent.<from>.<to>.ask
|
||||
// reply: cyrene.agent.<from>.<to>.reply.<askId>
|
||||
// event (to): cyrene.agent.<from>.<to>.event
|
||||
// event (bc): cyrene.agent.<from>.event
|
||||
//
|
||||
// The reply subject carries the ask id as its last token, giving each in-flight
|
||||
// ask a private reply address so concurrent asks from the same agent never
|
||||
// cross replies. To receive asks addressed to it, an agent subscribes to
|
||||
// cyrene.agent.*.<me>.ask. To receive events (both broadcasts and events
|
||||
// addressed to it) it subscribes to cyrene.agent.*.event and to
|
||||
// cyrene.agent.*.<me>.event.
|
||||
// 回复 subject 把 ask id 作为最后一个词元,给每个在途 ask 一个私有回复地址,因此同一
|
||||
// agent 的并发 ask 的回复绝不串线。要接收发给自己的 ask,agent 订阅
|
||||
// cyrene.agent.*.<me>.ask。要接收事件(广播与发给自己的事件),它订阅
|
||||
// cyrene.agent.*.event 与 cyrene.agent.*.<me>.event。
|
||||
//
|
||||
// # Delivery semantics
|
||||
// # 投递语义
|
||||
//
|
||||
// Ask blocks until a reply arrives, an ask-timeout elapses, or the passed
|
||||
// context is canceled; it returns ErrTimeout on the former and the wrapped
|
||||
// context error on the latter. Reply and the event senders are fire-and-forget
|
||||
// publishes that return the bus error (if any). Handler callbacks are invoked
|
||||
// on the bus's per-subscription worker goroutines, so an EventHandler may be
|
||||
// called concurrently from two subscriptions and must be safe for concurrent
|
||||
// use; an AskHandler for one agent is serialized by the bus.
|
||||
// Ask 阻塞到收到回复、ask 超时到期,或传入的上下文被取消;前一种情况返回
|
||||
// ErrTimeout,后一种情况返回包装后的上下文错误。Reply 与各事件发送方是即发即忘的
|
||||
// 发布,返回总线错误(若有)。处理回调在总线的按订阅工作 goroutine 上调用,因此一个
|
||||
// EventHandler 可能被两个订阅并发调用,必须可安全并发使用;而某个 agent 的
|
||||
// AskHandler 由总线串行化。
|
||||
package comm
|
||||
|
||||
import (
|
||||
@@ -45,82 +39,78 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// Kind is the wire kind of an agent-to-agent message.
|
||||
// Kind 是 agent 间消息的线上类别。
|
||||
type Kind string
|
||||
|
||||
const (
|
||||
// KindAsk is a question from one agent to another, answered by a reply.
|
||||
// KindAsk 是一个 agent 向另一个 agent 提的问题,由一条 reply 作答。
|
||||
KindAsk Kind = "ask"
|
||||
// KindReply is the answer to a previously published ask.
|
||||
// KindReply 是对此前发布的 ask 的应答。
|
||||
KindReply Kind = "reply"
|
||||
// KindEvent is a one-directional notification, broadcast or addressed.
|
||||
// KindEvent 是单向通知,可为广播或定向。
|
||||
KindEvent Kind = "event"
|
||||
)
|
||||
|
||||
// subjectPrefix is the fixed first two tokens of every agent-to-agent subject.
|
||||
// subjectPrefix 是每个 agent 间 subject 固定的前两个词元。
|
||||
const subjectPrefix = "cyrene.agent"
|
||||
|
||||
// DefaultAskTimeout caps how long Ask waits for a reply when the caller's
|
||||
// context carries no shorter deadline. It bounds a request/answer round so a
|
||||
// silent agent never hangs its caller.
|
||||
// DefaultAskTimeout 限制调用方上下文未携带更短期限时 Ask 等待回复的时长。它约束一个
|
||||
// 请求/应答往返,使沉默的 agent 永不吊死其调用方。
|
||||
const DefaultAskTimeout = 30 * time.Second
|
||||
|
||||
// Sentinel errors returned by the package. Use errors.Is to test for them; they
|
||||
// are returned directly or wrapped with %w so the underlying cause stays
|
||||
// available.
|
||||
// 本包返回的哨兵错误。用 errors.Is 判断它们;它们被直接返回或以 %w 包装返回,因此底层
|
||||
// 原因始终可得。
|
||||
var (
|
||||
// ErrTimeout is returned by Ask when no reply arrives before the ask
|
||||
// deadline (the caller's context deadline or DefaultAskTimeout) elapses.
|
||||
// ErrTimeout 在 ask 期限(调用方上下文期限或 DefaultAskTimeout)到期前没有回复
|
||||
// 到达时由 Ask 返回。
|
||||
ErrTimeout = errors.New("comm: reply timeout")
|
||||
|
||||
// ErrNoReply is returned by Ask when the ask cannot be delivered (the bus
|
||||
// rejected the publish/reply subscription), so no reply is expected.
|
||||
// ErrNoReply 在 ask 无法投递(总线拒绝了发布/回复订阅)时由 Ask 返回,此时不应
|
||||
// 期待任何回复。
|
||||
ErrNoReply = errors.New("comm: no reply")
|
||||
)
|
||||
|
||||
// Ask is a question sent from one agent to another.
|
||||
// Ask 是一个 agent 发给另一个 agent 的问题。
|
||||
type Ask struct {
|
||||
// ID uniquely identifies this ask within the answering exchange.
|
||||
// ID 在该问答交互内唯一标识本次 ask。
|
||||
ID string
|
||||
// From is the id of the asking agent (the sender).
|
||||
// From 是提问 agent(发送方)的 id。
|
||||
From string
|
||||
// To is the id of the agent that should answer (the receiver).
|
||||
// To 是应当作答的 agent(接收方)的 id。
|
||||
To string
|
||||
// Question is the human-readable question text.
|
||||
// Question 是人类可读的问题文本。
|
||||
Question string
|
||||
// Data carries an optional structured payload alongside the question.
|
||||
// Data 承载随问题一起发送的可选结构化载荷。
|
||||
Data any
|
||||
}
|
||||
|
||||
// Reply is the answer to a previously published ask.
|
||||
// Reply 是对此前发布的 ask 的应答。
|
||||
type Reply struct {
|
||||
// AskID is the ID of the ask this reply answers.
|
||||
// AskID 是本次回复所应答的 ask 的 ID。
|
||||
AskID string
|
||||
// From is the id of the replying agent (the sender).
|
||||
// From 是回复 agent(发送方)的 id。
|
||||
From string
|
||||
// To is the id of the asking agent (the receiver).
|
||||
// To 是提问 agent(接收方)的 id。
|
||||
To string
|
||||
// Text is the human-readable answer text.
|
||||
// Text 是人类可读的应答文本。
|
||||
Text string
|
||||
// Data carries an optional structured payload alongside the answer.
|
||||
// Data 承载随应答一起发送的可选结构化载荷。
|
||||
Data any
|
||||
}
|
||||
|
||||
// Event is a one-directional notification, either broadcast or addressed to one
|
||||
// agent. It has no reply.
|
||||
// Event 是单向通知,或广播,或定向给某个 agent。它没有回复。
|
||||
type Event struct {
|
||||
// From is the id of the emitting agent.
|
||||
// From 是发射事件的 agent 的 id。
|
||||
From string
|
||||
// Name is a short event name, e.g. "task.progress" or "tool.used".
|
||||
// Name 是简短的事件名,例如 "task.progress" 或 "tool.used"。
|
||||
Name string
|
||||
// Data carries an optional structured payload.
|
||||
// Data 承载可选的结构化载荷。
|
||||
Data any
|
||||
}
|
||||
|
||||
// token sanitizes an agent id so it can be used as a single subject token: any
|
||||
// rune outside [A-Za-z0-9_-] is replaced with "_". This keeps the id useful for
|
||||
// routing while guaranteeing dots (and the other subject metacharacters) never
|
||||
// leak into a token.
|
||||
// token 净化一个 agent id,使其可用作单个 subject 词元:凡不在 [A-Za-z0-9_-] 之内的
|
||||
// 字符都被替换为 "_"。这既让该 id 对路由仍然有用,又保证点号(以及其他 subject 元字符)
|
||||
// 绝不会漏进词元。
|
||||
func token(s string) string {
|
||||
var b strings.Builder
|
||||
for _, r := range s {
|
||||
@@ -134,42 +124,39 @@ func token(s string) string {
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// subjectFor builds the base subject for a from/to message of the given kind.
|
||||
// subjectFor 构建给定类别的 from/to 消息的基础 subject。
|
||||
func subjectFor(from, to string, kind Kind) string {
|
||||
return subjectPrefix + "." + token(from) + "." + token(to) + "." + string(kind)
|
||||
}
|
||||
|
||||
// askSubject is the subject an ask from→to is published to.
|
||||
// askSubject 是 from→to 的 ask 所发布到的 subject。
|
||||
func askSubject(from, to string) string {
|
||||
return subjectFor(from, to, KindAsk)
|
||||
}
|
||||
|
||||
// replySubject is the private subject a reply from→to answering askID is
|
||||
// published to. The trailing askID token gives each in-flight ask its own
|
||||
// address so replies never cross between concurrent asks.
|
||||
// replySubject 是应答 askID 的 from→to 回复所发布到的私有 subject。末尾的 askID 词元
|
||||
// 给每个在途 ask 自己的地址,因此并发 ask 之间回复绝不串线。
|
||||
func replySubject(from, to, askID string) string {
|
||||
return subjectPrefix + "." + token(from) + "." + token(to) + "." + string(KindReply) + "." + token(askID)
|
||||
}
|
||||
|
||||
// eventSubject is the subject of an event addressed from→to.
|
||||
// eventSubject 是 from→to 定向事件的 subject。
|
||||
func eventSubject(from, to string) string {
|
||||
return subjectFor(from, to, KindEvent)
|
||||
}
|
||||
|
||||
// broadcastEventSubject is the subject of a broadcast event emitted by from.
|
||||
// broadcastEventSubject 是 from 发射的广播事件的 subject。
|
||||
func broadcastEventSubject(from string) string {
|
||||
return subjectPrefix + "." + token(from) + "." + string(KindEvent)
|
||||
}
|
||||
|
||||
// askReceivePattern is the subscription pattern an agent named me uses to
|
||||
// receive asks addressed to it from any sender.
|
||||
// askReceivePattern 是名为 me 的 agent 用来接收任意发送方发给它的 ask 的订阅模式。
|
||||
func askReceivePattern(me string) string {
|
||||
return subjectPrefix + ".*." + token(me) + "." + string(KindAsk)
|
||||
}
|
||||
|
||||
// eventReceivePatterns are the subscription patterns an agent named me uses to
|
||||
// receive every event it should see: broadcasts from any sender and events
|
||||
// addressed specifically to it.
|
||||
// eventReceivePatterns 是名为 me 的 agent 用来接收它应当看到的所有事件的订阅模式:
|
||||
// 任意发送方的广播,以及专门定向给它的事件。
|
||||
func eventReceivePatterns(me string) []string {
|
||||
return []string{
|
||||
subjectPrefix + ".*." + string(KindEvent),
|
||||
|
||||
@@ -12,9 +12,8 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
)
|
||||
|
||||
// recvEvent blocks until an event arrives on ch or d elapses, failing the test
|
||||
// on timeout. It makes delivery assertions deterministic instead of relying on a
|
||||
// bare sleep.
|
||||
// recvEvent 阻塞到 ch 上到达一个事件或 d 到期,超时则判测试失败。它让投递断言确定化,
|
||||
// 而不必依赖干等的 sleep。
|
||||
func recvEvent(t *testing.T, ch <-chan Event, d time.Duration) Event {
|
||||
t.Helper()
|
||||
select {
|
||||
@@ -110,7 +109,7 @@ func TestAsk_TimeoutNoReply(t *testing.T) {
|
||||
b := bus.New()
|
||||
defer b.Close()
|
||||
alice := New(b, "alice")
|
||||
// bob has no ask handler, so nothing ever replies.
|
||||
// bob 没有任何 ask 处理函数,因此永远不会有人回复。
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
@@ -233,7 +232,7 @@ func TestEventSend_AddressedAndIsolated(t *testing.T) {
|
||||
t.Fatalf("bob event = %+v, want name=direct from=alice", e)
|
||||
}
|
||||
|
||||
// An event addressed to bob must not reach carol.
|
||||
// 定向给 bob 的事件不得抵达 carol。
|
||||
select {
|
||||
case e := <-carolEvents:
|
||||
t.Fatalf("carol unexpectedly received %+v", e)
|
||||
@@ -266,9 +265,8 @@ func TestEventHandler_Concurrent(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Fire a mix of broadcasts and events addressed to recv so both event
|
||||
// subscriptions (and their separate worker goroutines) invoke the handler
|
||||
// concurrently.
|
||||
// 混合发射广播与定向给 recv 的事件,使两个事件订阅(及其各自的工作 goroutine)
|
||||
// 并发调用处理函数。
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
@@ -349,7 +347,7 @@ func TestSend_NoErrorAndNoReceiver(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Publishing to a subject with no subscriber is a no-op, not an error.
|
||||
// 向没有订阅者的 subject 发布是空操作,不是错误。
|
||||
if err := alice.Send(ctx, "ghost", Event{Name: "nobody"}); err != nil {
|
||||
t.Fatalf("Send to lost receiver: %v", err)
|
||||
}
|
||||
|
||||
@@ -13,10 +13,8 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
)
|
||||
|
||||
// Probe is an agent-to-agent communication endpoint bound to a single agent id.
|
||||
// It is safe for concurrent use: the only mutable field is an atomic sequence
|
||||
// counter used to mint unique ask ids, and every in-flight ask carries its own
|
||||
// reply subscription and channel.
|
||||
// Probe 是绑定到单个 agent id 的 agent 间通信端点。它可安全并发使用:唯一可变字段是
|
||||
// 用于生成唯一 ask id 的原子序号计数器,且每个在途 ask 都带有自己的回复订阅与 channel。
|
||||
type Probe struct {
|
||||
bus bus.Bus
|
||||
agentID string
|
||||
@@ -25,23 +23,22 @@ type Probe struct {
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// Option customizes a Probe built by New.
|
||||
// Option 定制由 New 构建的 Probe。
|
||||
type Option func(*Probe)
|
||||
|
||||
// WithLogger replaces the probe's logger. The default is a discard logger so the
|
||||
// probe is quiet until configured.
|
||||
// WithLogger 替换探针的 logger。默认是丢弃式 logger,因此在配置之前探针是安静的。
|
||||
func WithLogger(l *slog.Logger) Option {
|
||||
return func(p *Probe) { p.log = newLogger(l) }
|
||||
}
|
||||
|
||||
// WithAskTimeout sets how long Ask waits for a reply when the caller's context
|
||||
// carries no shorter deadline. A non-positive value keeps DefaultAskTimeout.
|
||||
// WithAskTimeout 设置调用方上下文未携带更短期限时 Ask 等待回复的时长。非正值保持
|
||||
// DefaultAskTimeout。
|
||||
func WithAskTimeout(d time.Duration) Option {
|
||||
return func(p *Probe) { p.askTimeout = d }
|
||||
}
|
||||
|
||||
// New builds a Probe bound to agentID. The agentID is sanitized for use in bus
|
||||
// subjects (see token); an empty id panics because every probe must be routable.
|
||||
// New 构建一个绑定到 agentID 的 Probe。agentID 会被净化以便用于总线 subject
|
||||
// (见 token);空 id 会 panic,因为每个探针都必须可路由。
|
||||
func New(b bus.Bus, agentID string, opts ...Option) *Probe {
|
||||
if agentID == "" {
|
||||
panic("comm: New: agentID must not be empty")
|
||||
@@ -61,18 +58,15 @@ func New(b bus.Bus, agentID string, opts ...Option) *Probe {
|
||||
return p
|
||||
}
|
||||
|
||||
// AgentID returns the sanitized id this Probe is bound to.
|
||||
// AgentID 返回该 Probe 所绑定到的已净化 id。
|
||||
func (p *Probe) AgentID() string { return p.agentID }
|
||||
|
||||
// Ask sends a question to to and blocks until a matching Reply arrives, an ask
|
||||
// deadline elapses, or ctx is canceled. It returns the reply on success;
|
||||
// otherwise it returns ErrTimeout when the deadline elapsed without a reply,
|
||||
// ErrNoReply when the ask could not be delivered at all, or the wrapped context
|
||||
// error when ctx was canceled.
|
||||
// Ask 向 to 提一个问题并阻塞到匹配的 Reply 到达、ask 期限到期,或 ctx 被取消。成功时
|
||||
// 返回回复;否则期限到期而无回复时返回 ErrTimeout,ask 完全无法投递时返回 ErrNoReply,
|
||||
// ctx 被取消时返回包装后的上下文错误。
|
||||
//
|
||||
// The caller's context supplies a deadline if one is set; otherwise
|
||||
// p.askTimeout bounds the wait. The reply subscription is created before the
|
||||
// ask is published, so a fast answerer cannot be missed.
|
||||
// 调用方上下文设了期限就用它;否则由 p.askTimeout 限制等待时长。回复订阅先于 ask 发布
|
||||
// 创建,因此再快的应答方也不会被错过。
|
||||
func (p *Probe) Ask(ctx context.Context, to, question string, data any) (Reply, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return Reply{}, fmt.Errorf("comm: ask to %q: %w", to, err)
|
||||
@@ -120,9 +114,8 @@ func (p *Probe) Ask(ctx context.Context, to, question string, data any) (Reply,
|
||||
}
|
||||
}
|
||||
|
||||
// Reply publishes an answer to the ask identified by r.AskID, addressed to to.
|
||||
// The answerer defaults to this Probe's agent id when r.From is empty, so an
|
||||
// answering handler can simply pass the reply text and payload.
|
||||
// Reply 向 to 发布对 r.AskID 所标识 ask 的应答。r.From 为空时,应答方默认为本 Probe
|
||||
// 的 agent id,因此应答处理函数只需传入回复文本与载荷。
|
||||
func (p *Probe) Reply(ctx context.Context, to string, r Reply) error {
|
||||
if r.From == "" {
|
||||
r.From = p.agentID
|
||||
@@ -136,14 +129,12 @@ func (p *Probe) Reply(ctx context.Context, to string, r Reply) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Answer is a convenience for replying with plain text. It publishes a Reply
|
||||
// for askID to the asking agent to.
|
||||
// Answer 是以纯文本回复的便捷方式。它向提问 agent to 发布一条针对 askID 的 Reply。
|
||||
func (p *Probe) Answer(ctx context.Context, to, askID, text string, data any) error {
|
||||
return p.Reply(ctx, to, Reply{AskID: askID, From: p.agentID, To: to, Text: text, Data: data})
|
||||
}
|
||||
|
||||
// Send delivers an event addressed specifically to to. The emitter defaults to
|
||||
// this Probe's agent id when ev.From is empty.
|
||||
// Send 投递一个专门发给 to 的事件。ev.From 为空时,发射方默认为本 Probe 的 agent id。
|
||||
func (p *Probe) Send(ctx context.Context, to string, ev Event) error {
|
||||
if ev.From == "" {
|
||||
ev.From = p.agentID
|
||||
@@ -154,8 +145,8 @@ func (p *Probe) Send(ctx context.Context, to string, ev Event) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// PublishEvent broadcasts an event to every agent subscribed for events. The
|
||||
// emitter defaults to this Probe's agent id when ev.From is empty.
|
||||
// PublishEvent 向每个订阅了事件的 agent 广播一个事件。ev.From 为空时,发射方默认为本
|
||||
// Probe 的 agent id。
|
||||
func (p *Probe) PublishEvent(ctx context.Context, ev Event) error {
|
||||
if ev.From == "" {
|
||||
ev.From = p.agentID
|
||||
@@ -166,14 +157,13 @@ func (p *Probe) PublishEvent(ctx context.Context, ev Event) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// nextAskID mints a unique ask id from the agent id and an atomic counter. It is
|
||||
// safe for concurrent calls.
|
||||
// nextAskID 由 agent id 与一个原子计数器生成唯一的 ask id。可安全并发调用。
|
||||
func (p *Probe) nextAskID() string {
|
||||
return fmt.Sprintf("%s-%d", p.agentID, p.seq.Add(1))
|
||||
}
|
||||
|
||||
// newLogger returns l tagged with the orchestrations/comm source, swapping a nil
|
||||
// logger for a discard one so the probe can always log safely.
|
||||
// newLogger 返回标注了来源 orchestration/comm 的 l,并把 nil logger 换成丢弃式
|
||||
// logger,使探针总能安全记录日志。
|
||||
func newLogger(l *slog.Logger) *slog.Logger {
|
||||
if l == nil {
|
||||
l = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
|
||||
@@ -8,22 +8,18 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
)
|
||||
|
||||
// AskHandler processes an Ask addressed to this agent. It runs on the bus's
|
||||
// per-subscription worker goroutine; for a single agent asks are serialized. A
|
||||
// handler should answer via (p *Probe).Reply or Answer; a non-nil returned
|
||||
// error is logged by the probe and discarded.
|
||||
// AskHandler 处理发给本 agent 的 Ask。它运行在总线的按订阅工作 goroutine 上;对单个
|
||||
// agent 而言 ask 是串行的。处理函数应经 (p *Probe).Reply 或 Answer 作答;返回的非 nil
|
||||
// 错误由探针记日志并丢弃。
|
||||
type AskHandler func(ctx context.Context, a Ask) error
|
||||
|
||||
// EventHandler processes an Event addressed to, or broadcast to, this agent.
|
||||
// Because an agent subscribes to two event patterns (broadcast and addressed),
|
||||
// the bus may invoke a handler concurrently from two worker goroutines, so an
|
||||
// EventHandler must be safe for concurrent use.
|
||||
// EventHandler 处理定向给本 agent 或广播给本 agent 的 Event。由于一个 agent 订阅了两个
|
||||
// 事件模式(广播与定向),总线可能从两个工作 goroutine 并发调用同一个处理函数,因此
|
||||
// EventHandler 必须可安全并发使用。
|
||||
type EventHandler func(ctx context.Context, e Event) error
|
||||
|
||||
// SubscribeAsk registers h to receive every Ask addressed to this agent and
|
||||
// returns an unsubscribe function that removes the registration. When ctx is
|
||||
// canceled the registration is removed automatically; the returned function is
|
||||
// idempotent.
|
||||
// SubscribeAsk 登记 h 以接收发给本 agent 的每个 Ask,并返回一个解除登记的函数。ctx 被
|
||||
// 取消时登记自动移除;返回的函数是幂等的。
|
||||
func (p *Probe) SubscribeAsk(ctx context.Context, h AskHandler) (func(), error) {
|
||||
if h == nil {
|
||||
return nil, fmt.Errorf("comm: subscribe ask: handler must not be nil")
|
||||
@@ -45,14 +41,11 @@ func (p *Probe) SubscribeAsk(ctx context.Context, h AskHandler) (func(), error)
|
||||
return func() { unsub() }, nil
|
||||
}
|
||||
|
||||
// SubscribeEvent registers h to receive both broadcast events and events
|
||||
// addressed to this agent. It returns an unsubscribe function that removes both
|
||||
// registrations; it is idempotent and safe to call more than once. When ctx is
|
||||
// canceled both registrations are removed automatically.
|
||||
// SubscribeEvent 登记 h 以同时接收广播事件与定向给本 agent 的事件。它返回一个移除两处
|
||||
// 登记的函数;该函数幂等,多次调用也安全。ctx 被取消时两处登记都自动移除。
|
||||
//
|
||||
// The handler reaches the caller from two subscription worker goroutines and
|
||||
// may also run concurrently with a SubscribeAsk handler, so it must be safe for
|
||||
// concurrent use.
|
||||
// 处理函数会从两个订阅工作 goroutine 抵达调用方,也可能与 SubscribeAsk 的处理函数并发
|
||||
// 运行,因此它必须可安全并发使用。
|
||||
func (p *Probe) SubscribeEvent(ctx context.Context, h EventHandler) (func(), error) {
|
||||
if h == nil {
|
||||
return nil, fmt.Errorf("comm: subscribe event: handler must not be nil")
|
||||
|
||||
Reference in new issue
Block a user