Package io.agentscope.harness.agent.bus

Class WorkspaceMessageBus

java.lang.Object
io.agentscope.harness.agent.bus.WorkspaceMessageBus
All Implemented Interfaces:
MessageBus, AutoCloseable
public class WorkspaceMessageBus extends Object implements MessageBus
MessageBus implementation backed by AbstractFilesystem.

Works with any filesystem backend (local, remote, sandbox). Suitable for cross-process scenarios where multiple JVMs share the same workspace directory.

Mode D (pub/sub) is degraded to polling: subscribe(java.lang.String) returns a Flux that emits an empty signal every 3 seconds; publish(java.lang.String, java.util.Map<java.lang.String, java.lang.Object>) is a no-op.

File layout under busRoot: 

 {busRoot}/
   queues/{key-hash}/
     {entryId}.json       — Mode A entries (drain = ls + read + delete)
   logs/{key-hash}/
     {entryId}.json       — Mode C entries (append-only)

  • Constructor Details

    • WorkspaceMessageBus

      public WorkspaceMessageBus(AbstractFilesystem filesystem, String busRoot)
      Parameters:
      filesystem - any AbstractFilesystem implementation
      busRoot - absolute path within the filesystem for bus data (e.g. "/bus")

  • Method Details

    • queuePush

      public reactor.core.publisher.Mono<String> queuePush(String key, Map<String,Object> payload)
      Description copied from interface: MessageBus
      Append a payload to the drain queue at the given key.
      Specified by:
      queuePush in interface MessageBus
      Parameters:
      key - queue identifier (caller-defined naming convention)
      payload - JSON-serializable dict to enqueue
      Returns:
      the transport-level entry id

    • queueDrain

      public reactor.core.publisher.Mono<List<BusEntry>> queueDrain(String key, int maxCount)
      Description copied from interface: MessageBus
      Drain up to maxCount entries from the queue at the given key. Returned entries are removed from the queue atomically. A subsequent call returns only entries that arrived after this one.
      Specified by:
      queueDrain in interface MessageBus
      Parameters:
      key - queue identifier
      maxCount - maximum number of entries to return
      Returns:
      entries in arrival order; empty list when the queue is empty

    • queueDelete

      public reactor.core.publisher.Mono<Void> queueDelete(String key)
      Description copied from interface: MessageBus
      Delete the drain queue at the given key and all of its entries. Idempotent.
      Specified by:
      queueDelete in interface MessageBus
      Parameters:
      key - queue identifier

    • queuePeek

      public reactor.core.publisher.Mono<Boolean> queuePeek(String key)
      Description copied from interface: MessageBus
      Check whether the drain queue at the given key has any entries, without consuming them.
      Specified by:
      queuePeek in interface MessageBus
      Parameters:
      key - queue identifier
      Returns:
      true if the queue has at least one entry

    • logAppend

      public reactor.core.publisher.Mono<String> logAppend(String key, Map<String,Object> payload, int maxLen)
      Description copied from interface: MessageBus
      Append a payload to the replay log at the given key. Readers track their own cursor; entries persist until trimmed or capped by maxLen.
      Specified by:
      logAppend in interface MessageBus
      Parameters:
      key - log identifier
      payload - JSON-serializable dict to append
      maxLen - cap the log at approximately this many entries; older entries are trimmed when exceeded. Use 0 or negative for no cap.
      Returns:
      the transport-level entry id (usable as a cursor for subsequent reads)

    • logRead

      public reactor.core.publisher.Mono<List<BusEntry>> logRead(String key, String since, int maxCount)
      Description copied from interface: MessageBus
      Read up to maxCount entries from the replay log, starting after since. Non-destructive: the same entries can be returned to any number of readers.
      Specified by:
      logRead in interface MessageBus
      Parameters:
      key - log identifier
      since - cursor — return entries strictly newer than this id. null reads from the beginning.
      maxCount - maximum number of entries to return
      Returns:
      entries in append order; empty list when no entries are newer than since

    • logTrim

      public reactor.core.publisher.Mono<Void> logTrim(String key)
      Description copied from interface: MessageBus
      Delete the replay log at the given key and all of its entries. Idempotent.
      Specified by:
      logTrim in interface MessageBus
      Parameters:
      key - log identifier

    • publish

      public reactor.core.publisher.Mono<Void> publish(String key, Map<String,Object> payload)
      Description copied from interface: MessageBus
      Publish a payload on the broadcast channel. Only currently-subscribed listeners receive it.
      Specified by:
      publish in interface MessageBus
      Parameters:
      key - channel identifier
      payload - JSON-serializable dict

    • subscribe

      public reactor.core.publisher.Flux<Map<String,Object>> subscribe(String key)
      Description copied from interface: MessageBus
      Subscribe to a broadcast channel. Yields payloads published after subscription is established. The caller owns the subscription's lifetime — cancelling the Flux ends it.
      Specified by:
      subscribe in interface MessageBus
      Parameters:
      key - channel identifier
      Returns:
      stream of payloads