mq

Documentation

Index

• Variables
• func Close()
• func Topics() []*nsqd.Topic
• type Channel
  • func (p *Channel[M]) Do(fn Handler[M])
  • func (p *Channel[M]) Get() *Message
• type ChannelOption
• type Handler
• type Logger
  • func NewLogger() *Logger
  • func (l *Logger) Output(maxdepth int, s string) error
• type Message
• type Topic
  • func NewTopic[M any](name string) *Topic[M]
  • func (p *Topic[M]) DeferredPut(delay time.Duration, v M)
  • func (p *Topic[M]) DeleteExistingChannel(name string)
  • func (p *Topic[M]) Depth() int64
  • func (p *Topic[M]) GetOrCreateChannel(opt *ChannelOption) *Channel[M]
  • func (p *Topic[M]) MultiPut(vs []M)
  • func (p *Topic[M]) Put(v M)
  • func (p *Topic[M]) PutWithTimeout(v M, timeout time.Duration)
• type TopicOption

Variables

var (
	DefaultOptions = nsqd.NewOptions()
)

Functions

func Close

func Close()

func Topics

func Topics() []*nsqd.Topic

Types

type Channel

type Channel[M any] struct {
	Name string `json:"name,omitempty" yaml:"name,omitempty" toml:"name,omitempty" validate:"required"`
	// contains filtered or unexported fields
}

func (*Channel[M]) Do

func (p *Channel[M]) Do(fn Handler[M])

func (*Channel[M]) Get

func (p *Channel[M]) Get() *Message

type ChannelOption

type ChannelOption struct {
	Name string `json:"name,omitempty" yaml:"name,omitempty" toml:"name,omitempty" validate:"required"`

	// MaxAttempts 为 0 时，表示不限制重试次数
	MaxAttempts int64 `json:"max_attempts,omitempty" yaml:"max_attempts,omitempty" toml:"max_attempts,omitempty"`

	// Expire 为 0 时，表示不过期
	Expire time.Duration `json:"expire,omitempty" yaml:"expire,omitempty" toml:"expire,omitempty"`

	// MaxProcess 为 0 时，则限制为 1 个进程处理
	MaxProcess int64 `json:"max_process,omitempty" yaml:"max_process,omitempty" toml:"max_process,omitempty"`
}

type Handler

type Handler[M any] func(m *Message, v M) (err error)

type Logger

type Logger struct {
}

func NewLogger

func NewLogger() *Logger

func (*Logger) Output

func (l *Logger) Output(maxdepth int, s string) error

type Message

type Message struct {
	CreatedAt int64 `json:"created_at,omitempty" yaml:"created_at,omitempty" toml:"created_at,omitempty"`

	TraceId string `json:"trace_id,omitempty" yaml:"trace_id,omitempty" toml:"trace_id,omitempty"`

	// StartAt 为 0 时，表示立即开始
	StartAt int64 `json:"start_at,omitempty" yaml:"start_at,omitempty" toml:"start_at,omitempty"`

	// ExpireAt 为 0 时，表示不过期
	ExpireAt int64 `json:"expire_at,omitempty" yaml:"expire_at,omitempty" toml:"expire_at,omitempty"`

	Attempts uint16 `json:"-" yaml:"-" toml:"-"`

	MaxAttempts uint16 `json:"max_attempts,omitempty" yaml:"max_attempts,omitempty" toml:"max_attempts,omitempty"`

	Data []byte `json:"data,omitempty" yaml:"data,omitempty" toml:"data,omitempty"`
	// contains filtered or unexported fields
}

type Topic

type Topic[M any] struct {
	sync.RWMutex

	Name string `json:"name,omitempty" yaml:"name,omitempty" toml:"name,omitempty" validate:"required"`
	// contains filtered or unexported fields
}

func NewTopic

func NewTopic[M any](name string) *Topic[M]

func (*Topic[M]) DeferredPut

func (p *Topic[M]) DeferredPut(delay time.Duration, v M)

func (*Topic[M]) DeleteExistingChannel

func (p *Topic[M]) DeleteExistingChannel(name string)

func (*Topic[M]) Depth

func (p *Topic[M]) Depth() int64

func (*Topic[M]) GetOrCreateChannel

func (p *Topic[M]) GetOrCreateChannel(opt *ChannelOption) *Channel[M]

func (*Topic[M]) MultiPut

func (p *Topic[M]) MultiPut(vs []M)

func (*Topic[M]) Put

func (p *Topic[M]) Put(v M)

func (*Topic[M]) PutWithTimeout

func (p *Topic[M]) PutWithTimeout(v M, timeout time.Duration)

type TopicOption

type TopicOption struct {
	Name string `json:"name,omitempty" yaml:"name,omitempty" toml:"name,omitempty" validate:"required"`
}