broker

README

消息订阅 和 发布

Documentation

Index

• Variables
• func Logger() logger.ILogger
• func Register(name string, creator func(opts ...Option) IBroker)
• type Config
  • func NewConfig(opts ...Option) *Config
  • func (self *Config) Init(opts ...Option)
  • func (self *Config) Load() error
  • func (self *Config) Save(immed ...bool) error
  • func (self *Config) String() string
• type Handler
• type IBroker
  • func Default(opts ...Option) IBroker
  • func Use(name string, opts ...Option) IBroker
• type IEvent
• type ISubscriber
• type Message
• type Option
  • func Addrs(addrs ...string) Option
  • func ErrorHandler(h Handler) Option
  • func WithCodec(c codec.SerializeType) Option
  • func WithConfigPrefixName(prefixName string) Option
  • func WithContext(ctx context.Context) Option
  • func WithName(name string) Option
  • func WithRegisterInterval(t time.Duration) Option
  • func WithRegisterTTL(t time.Duration) Option
  • func WithRegistry(r registry.IRegistry) Option
  • func WithSecure(b bool) Option
  • func WithTLSConfig(t *tls.Config) Option
• type PublishConfig
• type PublishOption
  • func PublishContext(ctx context.Context) PublishOption
  • func WithPublishCodec(c codec.SerializeType) PublishOption
• type SubscribeConfig
  • func NewSubscribeOptions(opts ...SubscribeOption) *SubscribeConfig
• type SubscribeOption
  • func DisableAutoAck() SubscribeOption
  • func Queue(name string) SubscribeOption
  • func SubscribeContext(ctx context.Context) SubscribeOption

Variables

var (
	DefaultAddress = "127.0.0.1:0"
)

Functions

func Logger

func Logger() logger.ILogger

func Register

func Register(name string, creator func(opts ...Option) IBroker)

Types

type Config

type Config struct {
	config.Config `field:"-"`
	ErrorHandler  Handler            `field:"-"`
	Codec         codec.ICodec       `field:"-"` // Codec
	Registry      registry.IRegistry `field:"-"` // Registry used for clustering
	Context       context.Context    `field:"-"`
	TLSConfig     *tls.Config        `field:"-"`
	PrefixName    string             `field:"-"`
	Name          string             `field:"-"`
	Addrs         []string

	// Handler executed when error happens in broker mesage
	// processing
	Secure bool

	BroadcastVersion string
	RegisterTTL      time.Duration `field:"register_ttl"`
	RegisterInterval time.Duration
}

func NewConfig

func NewConfig(opts ...Option) *Config

func (*Config) Init

func (self *Config) Init(opts ...Option)

func (*Config) Load

func (self *Config) Load() error

func (*Config) Save

func (self *Config) Save(immed ...bool) error

func (*Config) String

func (self *Config) String() string

type Handler

type Handler func(IEvent) error

Handler is used to process messages via a subscription of a topic. The handler is passed a publication interface which contains the message and optional Ack method to acknowledge receipt of the message.

type IBroker

type IBroker interface {
	String() string // 返回对象名称
	Init(...Option) error
	Config() *Config
	Address() string
	Start() error // 开始阻塞监听
	Close() error
	Publish(topic string, m *Message, opts ...PublishOption) error
	Subscribe(topic string, h Handler, opts ...SubscribeOption) (ISubscriber, error)
}

IBroker 提供微服务之前的信息交换总线

func Default

func Default(opts ...Option) IBroker

func Use

func Use(name string, opts ...Option) IBroker

type IEvent

type IEvent interface {
	Topic() string
	Message() *Message
	Ack() error
	Error() error
}

Event is given to a subscription handler for processing

type ISubscriber

type ISubscriber interface {
	Config() *SubscribeConfig
	Topic() string
	Unsubscribe() error
}

Subscriber is a convenience return type for the Subscribe method

type Message

type Message struct {
	Header map[string]string
	Body   []byte
}

Message is a message send/received from the broker.

type Option

type Option func(*Config)

func Addrs

func Addrs(addrs ...string) Option

Addrs sets the host addresses to be used by the broker.

func ErrorHandler

func ErrorHandler(h Handler) Option

ErrorHandler will catch all broker errors that cant be handled in normal way, for example Codec errors.

func WithCodec

func WithCodec(c codec.SerializeType) Option

Codec sets the codec used for encoding/decoding used where a broker does not support headers.

func WithConfigPrefixName

func WithConfigPrefixName(prefixName string) Option

修改Config.json的路径

func WithContext

func WithContext(ctx context.Context) Option

func WithName

func WithName(name string) Option

修改Config.json的路径

func WithRegisterInterval

func WithRegisterInterval(t time.Duration) Option

func WithRegisterTTL

func WithRegisterTTL(t time.Duration) Option

func WithRegistry

func WithRegistry(r registry.IRegistry) Option

func WithSecure

func WithSecure(b bool) Option

Secure communication with the broker.

func WithTLSConfig

func WithTLSConfig(t *tls.Config) Option

Specify TLS Config.

type PublishConfig

type PublishConfig struct {
	Codec         codec.ICodec
	SerializeType codec.SerializeType
	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type PublishOption

type PublishOption func(*PublishConfig)

func PublishContext

func PublishContext(ctx context.Context) PublishOption

PublishContext set context.

func WithPublishCodec

func WithPublishCodec(c codec.SerializeType) PublishOption

type SubscribeConfig

type SubscribeConfig struct {
	// AutoAck defaults to true. When a handler returns
	// with a nil error the message is acked.
	AutoAck bool
	// Subscribers with the same queue name
	// will create a shared subscription where each
	// receives a subset of messages.
	Queue string

	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

func NewSubscribeOptions

func NewSubscribeOptions(opts ...SubscribeOption) *SubscribeConfig

type SubscribeOption

type SubscribeOption func(*SubscribeConfig)

func DisableAutoAck

func DisableAutoAck() SubscribeOption

DisableAutoAck will disable auto acking of messages after they have been handled.

func Queue

func Queue(name string) SubscribeOption

Queue sets the name of the queue to share messages on.

func SubscribeContext

func SubscribeContext(ctx context.Context) SubscribeOption

SubscribeContext set context.