bus

Documentation

Index

• Variables
• type ErrorHandler
• type InboundMessage
• type MediaRef
• type MessageBus
  • func New(bufferSize int) *MessageBus
  • func (mb *MessageBus) Close() error
  • func (mb *MessageBus) GetMetrics() Metrics
  • func (mb *MessageBus) PublishInbound(ctx context.Context, msg InboundMessage) error
  • func (mb *MessageBus) PublishOutbound(ctx context.Context, msg OutboundMessage) error
  • func (mb *MessageBus) PublishOutboundMedia(ctx context.Context, msg OutboundMediaMessage) error
  • func (mb *MessageBus) Request(ctx context.Context, msg InboundMessage, timeout time.Duration) (*OutboundMessage, error)
  • func (mb *MessageBus) SetErrorHandler(fn ErrorHandler)
  • func (mb *MessageBus) Subscribe(sub Subscription) string
  • func (mb *MessageBus) Unsubscribe(id string)
• type MessageHandler
• type MessageType
  • func (mt MessageType) String() string
• type Metrics
• type OutboundMediaMessage
• type OutboundMessage
• type Subscription

Variables

var (
	ErrBusClosed = errors.New("message bus is closed")
	ErrTimeout   = errors.New("request timed out")
)

Types

type ErrorHandler

type ErrorHandler func(err error)

ErrorHandler is called when an error occurs during dispatch.

type InboundMessage

type InboundMessage struct {
	ID        string
	Channel   string // "cli", "grpc", "system"
	SenderID  string
	ChatID    string
	Content   string
	Media     []MediaRef
	Metadata  map[string]string
	Timestamp time.Time
	Type      MessageType
}

InboundMessage represents a message coming into the agent from any channel.

type MediaRef

type MediaRef struct {
	Type     string // "image", "file", "audio"
	URL      string
	MimeType string
	Data     []byte
}

MediaRef references media in an inbound message.

type MessageBus

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

MessageBus is the central event bus for decoupling agent I/O.

func New

func New(bufferSize int) *MessageBus

New creates a new MessageBus.

func (*MessageBus) Close

func (mb *MessageBus) Close() error

Close shuts down the message bus.

func (*MessageBus) GetMetrics

func (mb *MessageBus) GetMetrics() Metrics

GetMetrics returns current bus metrics.

func (*MessageBus) PublishInbound

func (mb *MessageBus) PublishInbound(ctx context.Context, msg InboundMessage) error

PublishInbound dispatches an inbound message to matching subscribers.

func (*MessageBus) PublishOutbound

func (mb *MessageBus) PublishOutbound(ctx context.Context, msg OutboundMessage) error

PublishOutbound dispatches an outbound message to matching subscribers.

func (*MessageBus) PublishOutboundMedia

func (mb *MessageBus) PublishOutboundMedia(ctx context.Context, msg OutboundMediaMessage) error

PublishOutboundMedia dispatches a media message.

func (*MessageBus) Request

func (mb *MessageBus) Request(ctx context.Context, msg InboundMessage, timeout time.Duration) (*OutboundMessage, error)

Request sends an inbound message and waits for a correlated outbound reply.

func (*MessageBus) SetErrorHandler

func (mb *MessageBus) SetErrorHandler(fn ErrorHandler)

SetErrorHandler sets the callback for dispatch errors.

func (*MessageBus) Subscribe

func (mb *MessageBus) Subscribe(sub Subscription) string

Subscribe registers a handler and returns a subscription ID.

func (*MessageBus) Unsubscribe

func (mb *MessageBus) Unsubscribe(id string)

Unsubscribe removes a subscriber by ID.

type MessageHandler

type MessageHandler func(msg interface{})

MessageHandler is invoked when a message arrives.

type MessageType

type MessageType int

MessageType classifies the message for routing.

const (
	MessageTypeChat MessageType = iota
	MessageTypeToolCall
	MessageTypeToolResult
	MessageTypeAgentCall
	MessageTypeAgentResult
	MessageTypeSystem
	MessageTypeError
)

func (MessageType) String

func (mt MessageType) String() string

String returns the human-readable name.

type Metrics

type Metrics struct {
	Published int64
	Delivered int64
	Dropped   int64
}

Metrics tracks bus activity counters.

type OutboundMediaMessage

type OutboundMediaMessage struct {
	OutboundMessage
	MediaType string // "image", "file", "audio"
	MediaData []byte
	MediaURL  string
	FileName  string
}

OutboundMediaMessage carries binary/media payloads.

type OutboundMessage

type OutboundMessage struct {
	ID        string
	Channel   string
	ChatID    string
	Content   string
	ReplyToID string
	Metadata  map[string]string
	Timestamp time.Time
	Type      MessageType
	Streaming bool
}

OutboundMessage represents a message going from the agent to a channel.

type Subscription

type Subscription struct {
	ID          string
	Handler     MessageHandler
	MessageType *MessageType // nil = catch-all
	Channel     string       // empty = all channels
}

Subscription tracks a subscriber.