queue
The queue package wraps NSQ consumers to help reduce the amount of boilerplate
that each program has to implement to get rolling. It does not attempt to abstract NSQ away entirely.
View the go-nsq documentation for configuration options.
Example
c := queue.NewConsumer("events", "ingestion")
c.Set("nsqd", ":5001")
c.Set("concurrency", 15)
c.Set("max_attempts", 10)
c.Set("max_in_flight", 150)
c.Set("default_requeue_delay", "15s")
c.Start(nsq.HandlerFunc(func(msg *nsq.Message) error {
// do something
return nil
}))Usage
type Consumer
type Consumer struct {}Consumer convenience layer.
func NewConsumer
func NewConsumer(topic, channel string) *ConsumerNewConsumer returns a new consumer of topic and channel.
func (*Consumer) Set
func (c *Consumer) Set(option string, value interface{})Set option to value, any error will be returned in .Start().
Custom options implemented:
- `topic` consumer topic
- `channel` consumer channel
- `nsqd` nsqd address
- `nsqds` nsqd addresses
- `nsqlookupd` nsqlookupd address
- `nsqlookupds` nsqlookupd addresses
- `concurrency` concurrent handlers [1]
func (*Consumer) SetLogger
func (c *Consumer) SetLogger(log logger, level nsq.LogLevel)SetLogger replaces the default logger.
func (*Consumer) Start
func (c *Consumer) Start(handler nsq.Handler) errorStart consumer with handler.
func (*Consumer) Stop
func (c *Consumer) Stop() errorStop and wait.
License
MIT