nsqclient

README

nsqclient

nsq client for golang

Documentation

Index

• Variables
• func Client(name ...string) (pool.Pool, bool)
• func Connect(configs map[string]Config)
• func NewDebugError(text string) error
• func NewProducer(addr string) (*nsq.Producer, error)
• func NewProducerPool(addr string, initSize, maxSize int) (pool.Pool, error)
• func Publish(topic string, body []byte, nsqName ...string) error
• func Register(h INsqHandler, group string)
• func Run(group string, ctx context.Context, conf Config)
• type Config
• type HandleFunc
• type INsqHandler
• type NsqConsumer
  • func NewNsqConsumer(ctx context.Context, topic, channel string, options ...func(*nsq.Config)) (*NsqConsumer, error)
  • func (n *NsqConsumer) AddHandler(handler nsq.Handler)
  • func (n *NsqConsumer) Run(conf Config)
• type NsqHandler
  • func NewNsqHandler(options ...func(*NsqHandler)) *NsqHandler
  • func (h *NsqHandler) GetChannel() string
  • func (h *NsqHandler) GetChannelTopic() string
  • func (h *NsqHandler) GetHandle() HandleFunc
  • func (h *NsqHandler) GetMaxAttepts() uint16
  • func (h *NsqHandler) GetShouldRequeue(message *nsq.Message) (bool, time.Duration)
  • func (h *NsqHandler) GetSize() int
  • func (h *NsqHandler) GetTopic() string
  • func (h *NsqHandler) Group(group string)
  • func (h *NsqHandler) Init(fn func(ctx context.Context))
  • func (h *NsqHandler) IsOpenChannelTopic() bool
  • func (h *NsqHandler) RunInit(ctx context.Context)
  • func (h *NsqHandler) SetHandle(fn HandleFunc)
  • func (h *NsqHandler) SetShouldRequeue(fn func(message *nsq.Message) (bool, time.Duration))

Variables

var NsqGroups = make(map[string][]INsqHandler)

Functions

func Client

func Client(name ...string) (pool.Pool, bool)

func Connect

func Connect(configs map[string]Config)

func NewDebugError

func NewDebugError(text string) error

func NewProducer

func NewProducer(addr string) (*nsq.Producer, error)

CreateNSQProducer create nsq producer

func NewProducerPool

func NewProducerPool(addr string, initSize, maxSize int) (pool.Pool, error)

CreateNSQProducerPool create a nwq producer pool

func Publish

func Publish(topic string, body []byte, nsqName ...string) error

func Register

func Register(h INsqHandler, group string)

func Run

func Run(group string, ctx context.Context, conf Config)

Types

type Config

type Config struct {
	Host     string `json:"host"`
	Port     string `json:"port"`
	InitSize int    `toml:"init_size" json:"init_size"`
	MaxSize  int    `toml:"max_size" json:"max_size"`
}

type HandleFunc

type HandleFunc func(debug *debug.DebugTag, log logger.ILogger, message *nsq.Message) error

type INsqHandler

type INsqHandler interface {
	GetTopic() string
	GetChannel() string
	SetHandle(HandleFunc)
	GetHandle() HandleFunc
	SetShouldRequeue(fn func(message *nsq.Message) (bool, time.Duration))
	GetShouldRequeue(message *nsq.Message) (bool, time.Duration)
	GetMaxAttepts() uint16
	Group(group string)
	IsOpenChannelTopic() bool
	GetChannelTopic() string
	GetSize() int
	Init(fn func(ctx context.Context))
	RunInit(ctx context.Context)
}

type NsqConsumer

type NsqConsumer struct {
	Consumer *nsq.Consumer
	Handlers []nsq.Handler
	// contains filtered or unexported fields
}

func NewNsqConsumer

func NewNsqConsumer(ctx context.Context, topic, channel string, options ...func(*nsq.Config)) (*NsqConsumer, error)

func (*NsqConsumer) AddHandler

func (n *NsqConsumer) AddHandler(handler nsq.Handler)

func (*NsqConsumer) Run

func (n *NsqConsumer) Run(conf Config)

type NsqHandler

type NsqHandler struct {
	Topic            string
	Channel          string
	Size             int
	MaxAttepts       uint16
	OpenChannelTopic bool // 是否开启独立的topic [Topic.Channel]
	Handler          HandleFunc
	// contains filtered or unexported fields
}

func NewNsqHandler

func NewNsqHandler(options ...func(*NsqHandler)) *NsqHandler

func (*NsqHandler) GetChannel

func (h *NsqHandler) GetChannel() string

func (*NsqHandler) GetChannelTopic

func (h *NsqHandler) GetChannelTopic() string

func (*NsqHandler) GetHandle

func (h *NsqHandler) GetHandle() HandleFunc

func (*NsqHandler) GetMaxAttepts

func (h *NsqHandler) GetMaxAttepts() uint16

func (*NsqHandler) GetShouldRequeue

func (h *NsqHandler) GetShouldRequeue(message *nsq.Message) (bool, time.Duration)

func (*NsqHandler) GetSize

func (h *NsqHandler) GetSize() int

func (*NsqHandler) GetTopic

func (h *NsqHandler) GetTopic() string

func (*NsqHandler) Group

func (h *NsqHandler) Group(group string)

func (*NsqHandler) Init

func (h *NsqHandler) Init(fn func(ctx context.Context))

func (*NsqHandler) IsOpenChannelTopic

func (h *NsqHandler) IsOpenChannelTopic() bool

func (*NsqHandler) RunInit

func (h *NsqHandler) RunInit(ctx context.Context)

func (*NsqHandler) SetHandle

func (h *NsqHandler) SetHandle(fn HandleFunc)

func (*NsqHandler) SetShouldRequeue

func (h *NsqHandler) SetShouldRequeue(fn func(message *nsq.Message) (bool, time.Duration))