bus

package module

v0.28.0 Latest Latest Go to latest Published: May 20, 2026 License: MIT Imports: 11 Imported by: 4

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/infrago/bus

Links

Open Source Insights

README ¶

bus

bus 是 infrago 的模块。

包定位

类型：模块
作用：统一消息总线模块，负责发布/订阅、广播、异步投递。

主要功能

对上提供统一模块接口
对下通过驱动接口接入具体后端
支持按配置切换驱动实现

快速接入

import _ "github.com/infrago/bus"

[bus]
driver = "default"

驱动实现接口列表

以下接口由驱动实现（来自模块 driver.go）：

当前模块未提供独立驱动接口（或 driver.go 不存在）

全局配置项（所有配置键）

配置段：[bus]

driver
prefix
group
profile
weight
setting

说明

setting 一般用于向具体驱动透传专用参数
多实例配置请参考模块源码中的 Config/configure 处理逻辑
业务服务/消息名禁止使用保留前缀 _（该前缀用于 bus 内部控制通道）

Documentation ¶

Index ¶

func DispatchFinal(retries []time.Duration, attempt int) bool
func DispatchRetryDelay(retries []time.Duration, attempt int) (time.Duration, bool)
func IsRetryableDispatchError(err error) bool
func ListNodes() []infra.NodeInfo
func ListServices() []infra.ServiceInfo
func Stats() []infra.ServiceStats
type Config
type Configs
type Connection
type DispatchError
type Driver
type Handler
type Instance

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func DispatchFinal ¶ added in v0.13.0

func DispatchFinal(retries []time.Duration, attempt int) bool

func DispatchRetryDelay ¶ added in v0.13.0

func DispatchRetryDelay(retries []time.Duration, attempt int) (time.Duration, bool)

func IsRetryableDispatchError ¶ added in v0.13.0

func IsRetryableDispatchError(err error) bool

func ListNodes ¶ added in v0.7.0

func ListNodes() []infra.NodeInfo

func ListServices ¶ added in v0.7.0

func ListServices() []infra.ServiceInfo

func Stats ¶ added in v0.7.0

func Stats() []infra.ServiceStats

Types ¶

type Config ¶

type Config struct {
	Driver  string
	Weight  int
	Prefix  string
	Group   string
	Setting base.Map
}

type Configs ¶

type Configs map[string]Config

type Connection ¶ added in v0.7.0

type Connection interface {
	Open() error
	Close() error
	Start() error
	Stop() error

	RegisterService(subject string, retries []time.Duration) error
	RegisterMessage(subject string) error

	Request(subject string, data []byte, timeout time.Duration) ([]byte, error)
	Publish(subject string, data []byte) error
	Enqueue(subject string, data []byte) error

	Stats() []infra.ServiceStats
	ListNodes() []infra.NodeInfo
	ListServices() []infra.ServiceInfo
}

Connection defines a bus transport connection.

type DispatchError ¶ added in v0.13.0

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

func (DispatchError) Error ¶ added in v0.13.0

func (e DispatchError) Error() string

func (DispatchError) Result ¶ added in v0.13.0

func (e DispatchError) Result() base.Res

func (DispatchError) Retryable ¶ added in v0.13.0

func (e DispatchError) Retryable() bool

type Driver ¶

type Driver interface {
	Connect(*Instance) (Connection, error)
}

Driver connections a bus transport.

type Handler ¶

type Handler func([]byte) ([]byte, error)

Handler processes incoming payload and returns reply bytes for call.

type Instance ¶

type Instance struct {
	Name   string
	Config Config
}

func (*Instance) HandleAsync ¶ added in v0.7.0

func (inst *Instance) HandleAsync(data []byte) error

HandleAsync is kept as compatibility alias of HandleServiceAsync.

func (*Instance) HandleCall ¶ added in v0.7.0

func (inst *Instance) HandleCall(data []byte) ([]byte, error)

HandleCall handles request/reply for a bus instance.

func (*Instance) HandleMessage ¶ added in v0.13.0

func (inst *Instance) HandleMessage(data []byte) error

HandleMessage handles async execution (broadcast/rolecast) for message entries.

func (*Instance) HandleServiceAsync ¶ added in v0.13.0

func (inst *Instance) HandleServiceAsync(data []byte, attempt int, final bool) error

HandleServiceAsync handles async execution (queue) for service entries.

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL