接口interface io.agentscope.harness.agent.bus 2.x

MessageBus

public interface MessageBus extends AutoCloseable

源码Source io/agentscope/harness/agent/bus/MessageBus.java:45–267

定义消息总线契约的接口。位于 io.agentscope.harness.agent.bus 包，实现 AutoCloseable。核心成员包括 queuePush、queueDrain、queueDelete、queuePeek、inboxHasMessages、logAppend。

Live message transport for coordinating work across sessions and processes.

Three consumption modes are exposed:

Drain queue (Mode A) — single-consumer, ack-on-read. Each entry is returned at most once; storage drops it the moment it is read.
Replay log (Mode C) — multi-consumer, externally bounded. Each reader tracks its own cursor; entries persist until trimmed or capped by maxLen.
Transient broadcast (Mode D) — fire-and-forget pub/sub. Only currently-subscribed listeners receive a payload; no history is retained.

Domain helpers (#inboxPush, #inboxDrain, #enqueueWakeup) map higher-level concepts onto these primitives with a fixed key-naming convention.

Implementations:

WorkspaceMessageBus — single-process, no external dependencies

字段摘要Field Summary

字段Field	类型Type	说明Description
`SESSION_EVENTS_KEY_PREFIX`	String	会话事件键 PREFIX字段。 The SESSION EVENTS KEY PREFIX field.
`SESSION_REPLAY_MAX_LEN`	int	会话 REPLAY 最大 LEN字段。 The SESSION REPLAY MAX LEN field.

方法摘要Method Summary

方法Method	说明Description
`public Mono<String> queuePush(String key, Map<String, Object> payload)`	执行队列推送相关操作。 Append a payload to the drain queue at the given key.
`public Mono<List<BusEntry>> queueDrain(String key, int maxCount)`	执行 `queueDrain` 方法。 Drain up to `maxCount` entries from the queue at the given key.
`public Mono<Void> queueDelete(String key)`	设置队列删除。 Delete the drain queue at the given key and all of its entries.
`public Mono<Boolean> queuePeek(String key)`	设置该参数。 Check whether the drain queue at the given key has any entries, without consuming them.
`default public Mono<Boolean> inboxHasMessages(String sessionId)` default	设置收件箱有消息。 Check whether a session's inbox has pending messages without consuming them.
`public Mono<String> logAppend(String key, Map<String, Object> payload, int maxLen)`	执行日志追加相关操作。 Append a payload to the replay log at the given key.
`public Mono<List<BusEntry>> logRead(String key, String since, int maxCount)`	执行日志读取相关操作。 Read up to `maxCount` entries from the replay log, starting after `since`.
`public Mono<Void> logTrim(String key)`	设置日志裁剪。 Delete the replay log at the given key and all of its entries.
`public Mono<Void> publish(String key, Map<String, Object> payload)`	执行发布相关操作。 Publish a payload on the broadcast channel.
`public Flux<Map<String, Object>> subscribe(String key)`	设置订阅。 Subscribe to a broadcast channel.
`default public Mono<Void> inboxPush(String sessionId, Map<String, Object> msg)` default	执行收件箱推送相关操作。 Push a message into a session's inbox queue.
`default public Mono<List<BusEntry>> inboxDrain(String sessionId, int maxCount)` default	执行 `inboxDrain` 方法。 Drain pending inbox messages for a session.
`default public Mono<Void> enqueueWakeup(String userId, String sessionId, String agentId)` default	执行 `enqueueWakeup` 方法。 Enqueue a wakeup request and signal dispatchers.
`default public Mono<Void> enqueueWakeup(String sessionId, String agentId)` default	执行 `enqueueWakeup` 方法。 Convenience overload without userId.
`default public Flux<Map<String, Object>> subscribeWakeup()` default	订阅 `subscribeWakeup`。 Subscribe to the shared wakeup signal channel.
`default public Mono<String> sessionPublishEvent(String sessionId, Map<String, Object> event)` default	执行会话发布事件相关操作。 Append a session event to the replay log and fan it out live.
`default public Mono<List<BusEntry>> sessionReadEvents(String sessionId, String since, int maxCount)` default	执行会话读取事件相关操作。 Read events from a session's replay log for catch-up / reconnection.
`default public Flux<Map<String, Object>> sessionSubscribeEvents(String sessionId)` default	设置会话订阅事件。 Live-subscribe to a session's published events.
`default public Mono<Void> sessionTrimEvents(String sessionId)` default	设置会话裁剪事件。 Trim the session's event replay log.
`default public void close()` default	关闭连接并释放相关资源。 Release underlying transport resources.

字段详情Field Detail

SESSION_EVENTS_KEY_PREFIX

String SESSION_EVENTS_KEY_PREFIX

会话事件键 PREFIX字段。声明为 String SESSION_EVENTS_KEY_PREFIX。

The SESSION EVENTS KEY PREFIX field.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:209

SESSION_REPLAY_MAX_LEN

int SESSION_REPLAY_MAX_LEN

会话 REPLAY 最大 LEN字段。声明为 int SESSION_REPLAY_MAX_LEN。

The SESSION REPLAY MAX LEN field.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:210

方法详情Method Detail

queuePush

public Mono<String> queuePush(String key, Map<String, Object> payload)

执行队列推送相关操作。

Append a payload to the drain queue at the given key.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。queue identifier (caller-defined naming convention)
`payload`	Map<String, Object>	负载（类型 `Map`）。JSON-serializable dict to enqueue

返回Returns Mono<String>

返回Mono。the transport-level entry id

源码Source io/agentscope/harness/agent/bus/MessageBus.java:56

queueDrain

public Mono<List<BusEntry>> queueDrain(String key, int maxCount)

执行 queueDrain 方法。

Drain up to maxCount entries from the queue at the given key.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。queue identifier
`maxCount`	int	最大数量（类型 `int`）。maximum number of entries to return

返回Returns Mono<List<BusEntry>>

返回Mono>。entries in arrival order; empty list when the queue is empty

源码Source io/agentscope/harness/agent/bus/MessageBus.java:67

queueDelete

public Mono<Void> queueDelete(String key)

设置队列删除。

Delete the drain queue at the given key and all of its entries.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。queue identifier

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:74

queuePeek

public Mono<Boolean> queuePeek(String key)

设置该参数。

Check whether the drain queue at the given key has any entries, without consuming them.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。queue identifier

返回Returns Mono<Boolean>

返回Mono。true if the queue has at least one entry

源码Source io/agentscope/harness/agent/bus/MessageBus.java:82

inboxHasMessages

default public Mono<Boolean> inboxHasMessages(String sessionId)

default

设置收件箱有消息。

Check whether a session's inbox has pending messages without consuming them.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话到检查。the session to check

返回Returns Mono<Boolean>

返回Mono。true if the inbox has at least one message

源码Source io/agentscope/harness/agent/bus/MessageBus.java:92–94

logAppend

public Mono<String> logAppend(String key, Map<String, Object> payload, int maxLen)

执行日志追加相关操作。

Append a payload to the replay log at the given key.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。log identifier
`payload`	Map<String, Object>	负载（类型 `Map`）。JSON-serializable dict to append
`maxLen`	int	最大 Len（类型 `int`）。cap the log at approximately this many entries; older entries are trimmed when exceeded. Use `0` or negative for no cap.

返回Returns Mono<String>

返回Mono。the transport-level entry id (usable as a cursor for subsequent reads)

源码Source io/agentscope/harness/agent/bus/MessageBus.java:108

logRead

public Mono<List<BusEntry>> logRead(String key, String since, int maxCount)

执行日志读取相关操作。

Read up to maxCount entries from the replay log, starting after since.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。log identifier
`since`	String	since（类型 `String`）。cursor — return entries strictly newer than this id. `null` reads from the beginning.
`maxCount`	int	最大数量（类型 `int`）。maximum number of entries to return

返回Returns Mono<List<BusEntry>>

返回Mono>。entries in append order; empty list when no entries are newer than `since`

源码Source io/agentscope/harness/agent/bus/MessageBus.java:120

logTrim

public Mono<Void> logTrim(String key)

设置日志裁剪。

Delete the replay log at the given key and all of its entries.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。log identifier

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:127

publish

public Mono<Void> publish(String key, Map<String, Object> payload)

执行发布相关操作。

Publish a payload on the broadcast channel.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。channel identifier
`payload`	Map<String, Object>	负载（类型 `Map`）。JSON-serializable dict

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:137

public Flux<Map<String, Object>> subscribe(String key)

设置订阅。

Subscribe to a broadcast channel.

参数Parameters

名称Name	类型Type	说明Description
`key`	String	键（类型 `String`）。channel identifier

返回Returns Flux<Map<String, Object>>

负载的流。stream of payloads

源码Source io/agentscope/harness/agent/bus/MessageBus.java:146

inboxPush

default public Mono<Void> inboxPush(String sessionId, Map<String, Object> msg)

default

执行收件箱推送相关操作。

Push a message into a session's inbox queue.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话标识（类型 `String`）。the recipient session
`msg`	Map<String, Object>	消息（类型 `Map`）。JSON-serializable payload (typically a serialized HintBlock)

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:156–158

inboxDrain

default public Mono<List<BusEntry>> inboxDrain(String sessionId, int maxCount)

default

执行 inboxDrain 方法。

Drain pending inbox messages for a session.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话标识（类型 `String`）。the session whose inbox to drain
`maxCount`	int	最大数量（类型 `int`）。maximum entries to drain

返回Returns Mono<List<BusEntry>>

返回Mono>。entries in arrival order

源码Source io/agentscope/harness/agent/bus/MessageBus.java:167–169

enqueueWakeup

default public Mono<Void> enqueueWakeup(String userId, String sessionId, String agentId)

default

执行 enqueueWakeup 方法。

Enqueue a wakeup request and signal dispatchers.

参数Parameters

名称Name	类型Type	说明Description
`userId`	String	用户标识（类型 `String`）。the owning user id (for multi-user isolation)
`sessionId`	String	会话标识（类型 `String`）。the session to wake
`agentId`	String	智能体标识（类型 `String`）。the agent that owns the session

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:180–188

enqueueWakeup

default public Mono<Void> enqueueWakeup(String sessionId, String agentId)

default

执行 enqueueWakeup 方法。

Convenience overload without userId.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话标识（类型 `String`）。The session id.
`agentId`	String	智能体标识（类型 `String`）。The agent id.

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:194–196

subscribeWakeup

default public Flux<Map<String, Object>> subscribeWakeup()

default

订阅 subscribeWakeup。

Subscribe to the shared wakeup signal channel.

返回Returns Flux<Map<String, Object>>

返回Flux>。stream of signal payloads (each indicates "drain the wakeup queue now")

源码Source io/agentscope/harness/agent/bus/MessageBus.java:203–205

sessionPublishEvent

default public Mono<String> sessionPublishEvent(String sessionId, Map<String, Object> event)

default

执行会话发布事件相关操作。

Append a session event to the replay log and fan it out live.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话标识（类型 `String`）。the session this event belongs to
`event`	Map<String, Object>	事件（类型 `Map`）。JSON-serializable event payload

返回Returns Mono<String>

返回Mono。the replay-log entry id

源码Source io/agentscope/harness/agent/bus/MessageBus.java:220–229

sessionReadEvents

default public Mono<List<BusEntry>> sessionReadEvents(String sessionId, String since, int maxCount)

default

执行会话读取事件相关操作。

Read events from a session's replay log for catch-up / reconnection.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话标识（类型 `String`）。the session whose events to read
`since`	String	since（类型 `String`）。cursor — return entries strictly newer than this id; `null` reads all
`maxCount`	int	最大数量（类型 `int`）。maximum events to return

返回Returns Mono<List<BusEntry>>

返回Mono>。entries in append order

源码Source io/agentscope/harness/agent/bus/MessageBus.java:239–241

sessionSubscribeEvents

default public Flux<Map<String, Object>> sessionSubscribeEvents(String sessionId)

default

设置会话订阅事件。

Live-subscribe to a session's published events.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话到订阅到。the session to subscribe to

返回Returns Flux<Map<String, Object>>

事件负载的流。stream of event payloads

源码Source io/agentscope/harness/agent/bus/MessageBus.java:250–252

sessionTrimEvents

default public Mono<Void> sessionTrimEvents(String sessionId)

default

设置会话裁剪事件。

Trim the session's event replay log.

参数Parameters

名称Name	类型Type	说明Description
`sessionId`	String	会话标识（类型 `String`）。the session whose event log to clear

返回Returns Mono<Void>

返回Mono。Returns mono.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:260–262

close

default public void close()

default

关闭连接并释放相关资源。

Release underlying transport resources.

源码Source io/agentscope/harness/agent/bus/MessageBus.java:265–266