nsqconsumer

README

Package nsqconsumer v1.2.0

Documentation

Index

• func NewError(err error, opts ErrorOpts) error
• type Consumer
  • func New(opts ConsumerOpts) (Consumer, error)
• type ConsumerOpts
• type Error
  • func (nsqerr Error) Error() string
• type ErrorOpts
• type NsqMessageDeserialize
  • func FromMessageSerialize(msg *nsqproducer.NsqMessageSerialize) *NsqMessageDeserialize
  • func (msg *NsqMessageDeserialize) TouchUntilClosed() chan<- struct{}

Functions

func NewError

func NewError(err error, opts ErrorOpts) error

Types

type Consumer

type Consumer interface {
	Start(ctx context.Context) func()
}

func New

func New(opts ConsumerOpts) (Consumer, error)

type ConsumerOpts

type ConsumerOpts struct {
	NsqConfig      *nsq.Config
	NsqLookupdURLs []string
	Topic          string
	Channel        string
	MaxInFlight    int
	SkipLogSet     map[string]bool
	// PostponeProducer is an NSQ producer user to send postponed messages
	PostponeProducer nsqproducer.Producer
	// How long can the consumer keep the message before the message is considered as 'Timed Out'
	MsgTimeout     time.Duration
	MessageHandler func(context.Context, *NsqMessageDeserialize) error
}

type Error

type Error struct {
	// contains filtered or unexported fields
}

func (Error) Error

func (nsqerr Error) Error() string

type ErrorOpts

type ErrorOpts struct {
	NoRetry bool
}

type NsqMessageDeserialize

type NsqMessageDeserialize struct {
	RequestID string          `json:"request_id"`
	Type      string          `json:"type"`
	At        int64           `json:"at"`
	Payload   json.RawMessage `json:"payload"`
	NsqMsg    *nsq.Message
}

func FromMessageSerialize

func FromMessageSerialize(msg *nsqproducer.NsqMessageSerialize) *NsqMessageDeserialize

FromMessageSerialize let you transform a Serialized message to a DeserializeMessage for a consumer Its use is mostly for testing as writing manual `json.RawMessage` is boring

func (*NsqMessageDeserialize) TouchUntilClosed

func (msg *NsqMessageDeserialize) TouchUntilClosed() chan<- struct{}

TouchUntilClosed returns a channel which has to be closed by the called Until the channel is closed, the NSQ message will be touched every 40 secs to ensure NSQ does not consider the message as failed because of time out.