mq

Documentation

Index

• Constants
• func CloneTopic() map[string]*nsqd.Topic
• func Close()
• func GetDepth() (depth int64)
• func GetTopic(name fmt.Stringer) *nsqd.Topic
• func GetTopicDepth(name fmt.Stringer) int64
• func RegisterHandel(topicName fmt.Stringer, channelName fmt.Stringer, handel *Handle)
• func RegisterTopic(name fmt.Stringer, messageType interface{})
• func Start(o *Option) error
• type Channel
  • func (p Channel) GoString() string
  • func (p Channel) String() string
• type Handle
• type HandleFunc
• type HandleReq
• type HandleRsp
• type Logger
  • func NewLogger() *Logger
  • func (l *Logger) Output(maxdepth int, s string) error
• type Option
  • func NewOption() *Option
• type PublishRsp
  • func Publish(topicName fmt.Stringer, message interface{}) (*PublishRsp, error)
• type Topic
  • func (p Topic) GoString() string
  • func (p Topic) String() string

Constants

const (
	Version = 1
)

Functions

func CloneTopic

func CloneTopic() map[string]*nsqd.Topic

func Close

func Close()

func GetDepth

func GetDepth() (depth int64)

func GetTopic

func GetTopic(name fmt.Stringer) *nsqd.Topic

func GetTopicDepth

func GetTopicDepth(name fmt.Stringer) int64

func RegisterHandel

func RegisterHandel(topicName fmt.Stringer, channelName fmt.Stringer, handel *Handle)

func RegisterTopic

func RegisterTopic(name fmt.Stringer, messageType interface{})

func Start

func Start(o *Option) error

Types

type Channel

type Channel string

func (Channel) GoString

func (p Channel) GoString() string

func (Channel) String

func (p Channel) String() string

type Handle

type Handle struct {
	HandleFunc HandleFunc `json:"handle_func,omitempty"`
	// 默认为无限重试
	MaxAttempts uint16 `json:"max_attempts,omitempty"`

	MaxProcessCnt int64 `json:"max_process_cnt,omitempty"`

	// 消息超时时间
	MsgTimeout time.Duration `json:"msg_timeout,omitempty"`
}

type HandleFunc

type HandleFunc func(msg *HandleReq) (*HandleRsp, error)

type HandleReq

type HandleReq struct {
	MsgId    string    `json:"msg_id,omitempty"`
	Attempts uint16    `json:"attempts,omitempty"`
	PubAt    time.Time `json:"pub_at,omitempty"`

	Body    []byte      `json:"body,omitempty"`
	Message interface{} `json:"message,omitempty"`
}

type HandleRsp

type HandleRsp struct {
	NeedRetry    bool `json:"need_retry,omitempty"`
	SkipRetryCnt bool `json:"skip_retry_cnt,omitempty"`

	WaitTime time.Duration `json:"wait_time,omitempty"`
}

type Logger

type Logger struct {
}

func NewLogger

func NewLogger() *Logger

func (*Logger) Output

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

type Option

type Option struct {
	// 数据存储的路径
	DataPath string `json:"data_path,omitempty"`
	// 内存队列的大小
	MemQueueSize int64 `json:"mem_queue_size,omitempty"`
}

func NewOption

func NewOption() *Option

type PublishRsp

type PublishRsp struct {
	Version uint32 `json:"version,omitempty"`
	MsgId   string `json:"msgid,omitempty"`
}

func Publish

func Publish(topicName fmt.Stringer, message interface{}) (*PublishRsp, error)

type Topic

type Topic string

func (Topic) GoString

func (p Topic) GoString() string

func (Topic) String

func (p Topic) String() string