func (r *Consumer) lookupdLoop() {
	r.rngMtx.Lock()
	jitter := time.Duration(int64(r.rng.Float64() *
		r.config.LookupdPollJitter * float64(r.config.LookupdPollInterval)))
	r.rngMtx.Unlock()
	var ticker *time.Ticker
	select {
	case <-time.After(jitter):
	case <-r.exitChan:
		goto exit
	}
	ticker = time.NewTicker(r.config.LookupdPollInterval)
	for {
		select {
		case <-ticker.C:
			r.queryLookupd()
		case <-r.lookupdRecheckChan:
			r.queryLookupd()
		case <-r.exitChan:
			goto exit
		}
	}
exit:
	if ticker != nil {
		ticker.Stop()
	}
	r.log(LogLevelInfo, "exiting lookupdLoop")
	r.wg.Done()
}
poll all known lookup servers every LookupdPollInterval
package nsq
import (
	"bytes"
	"errors"
	"fmt"
	"log"
	"math"
	"math/rand"
	"net"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)
type Handler interface {
	HandleMessage(message *Message) error
}
type HandlerFunc func(message *Message) error
func (h HandlerFunc) HandleMessage(m *Message) error {
	return h(m)
}
type DiscoveryFilter interface {
	Filter([]string) []string
}
type FailedMessageLogger interface {
	LogFailedMessage(message *Message)
}
type ConsumerStats struct {
	MessagesReceived uint64
	MessagesFinished uint64
	MessagesRequeued uint64
	Connections      int
}
var instCount int64
type backoffSignal int
const (
	backoffFlag backoffSignal = iota
	continueFlag
	resumeFlag
)
type Consumer struct {
	messagesReceived uint64
	messagesFinished uint64
	messagesRequeued uint64
	totalRdyCount    int64
	backoffDuration  int64
	backoffCounter   int32
	maxInFlight      int32
	mtx sync.RWMutex
	logger   logger
	logLvl   LogLevel
	logGuard sync.RWMutex
	behaviorDelegate interface{}
	id      int64
	topic   string
	channel string
	config  Config
	rngMtx sync.Mutex
	rng    *rand.Rand
	needRDYRedistributed int32
	backoffMtx sync.RWMutex
	incomingMessages chan *Message
	rdyRetryMtx    sync.RWMutex
	rdyRetryTimers map[string]*time.Timer
	pendingConnections map[string]*Conn
	connections        map[string]*Conn
	nsqdTCPAddrs []string
	lookupdRecheckChan chan int
	lookupdHTTPAddrs   []string
	lookupdQueryIndex  int
	wg              sync.WaitGroup
	runningHandlers int32
	stopFlag        int32
	connectedFlag   int32
	stopHandler     sync.Once
	exitHandler     sync.Once
	StopChan chan int
	exitChan chan int
}
func NewConsumer(topic string, channel string, config *Config) (*Consumer, error) {
	config.assertInitialized()
	if err := config.Validate(); err != nil {
		return nil, err
	}
	if !IsValidTopicName(topic) {
		return nil, errors.New("invalid topic name")
	}
	if !IsValidChannelName(channel) {
		return nil, errors.New("invalid channel name")
	}
	r := &Consumer{
		id: atomic.AddInt64(&instCount, 1),
		topic:   topic,
		channel: channel,
		config:  *config,
		logger:      log.New(os.Stderr, "", log.Flags()),
		logLvl:      LogLevelInfo,
		maxInFlight: int32(config.MaxInFlight),
		incomingMessages: make(chan *Message),
		rdyRetryTimers:     make(map[string]*time.Timer),
		pendingConnections: make(map[string]*Conn),
		connections:        make(map[string]*Conn),
		lookupdRecheckChan: make(chan int, 1),
		rng: rand.New(rand.NewSource(time.Now().UnixNano())),
		StopChan: make(chan int),
		exitChan: make(chan int),
	}
	r.wg.Add(1)
	go r.rdyLoop()
	return r, nil
}
func (r *Consumer) Stats() *ConsumerStats {
	return &ConsumerStats{
		MessagesReceived: atomic.LoadUint64(&r.messagesReceived),
		MessagesFinished: atomic.LoadUint64(&r.messagesFinished),
		MessagesRequeued: atomic.LoadUint64(&r.messagesRequeued),
		Connections:      len(r.conns()),
	}
}
func (r *Consumer) conns() []*Conn {
	r.mtx.RLock()
	conns := make([]*Conn, 0, len(r.connections))
	for _, c := range r.connections {
		conns = append(conns, c)
	}
	r.mtx.RUnlock()
	return conns
}
func (r *Consumer) SetLogger(l logger, lvl LogLevel) {
	r.logGuard.Lock()
	defer r.logGuard.Unlock()
	r.logger = l
	r.logLvl = lvl
}
func (r *Consumer) getLogger() (logger, LogLevel) {
	r.logGuard.RLock()
	defer r.logGuard.RUnlock()
	return r.logger, r.logLvl
}
func (r *Consumer) SetBehaviorDelegate(cb interface{}) {
	matched := false
	if _, ok := cb.(DiscoveryFilter); ok {
		matched = true
	}
	if !matched {
		panic("behavior delegate does not have any recognized methods")
	}
	r.behaviorDelegate = cb
}
func (r *Consumer) perConnMaxInFlight() int64 {
	b := float64(r.getMaxInFlight())
	s := b / float64(len(r.conns()))
	return int64(math.Min(math.Max(1, s), b))
}
func (r *Consumer) IsStarved() bool {
	for _, conn := range r.conns() {
		threshold := int64(float64(atomic.LoadInt64(&conn.lastRdyCount)) * 0.85)
		inFlight := atomic.LoadInt64(&conn.messagesInFlight)
		if inFlight >= threshold && inFlight > 0 && !conn.IsClosing() {
			return true
		}
	}
	return false
}
func (r *Consumer) getMaxInFlight() int32 {
	return atomic.LoadInt32(&r.maxInFlight)
}
func (r *Consumer) ChangeMaxInFlight(maxInFlight int) {
	if r.getMaxInFlight() == int32(maxInFlight) {
		return
	}
	atomic.StoreInt32(&r.maxInFlight, int32(maxInFlight))
	for _, c := range r.conns() {
		r.maybeUpdateRDY(c)
	}
}
func (r *Consumer) ConnectToNSQLookupd(addr string) error {
	if atomic.LoadInt32(&r.stopFlag) == 1 {
		return errors.New("consumer stopped")
	}
	if atomic.LoadInt32(&r.runningHandlers) == 0 {
		return errors.New("no handlers")
	}
	if err := validatedLookupAddr(addr); err != nil {
		return err
	}
	atomic.StoreInt32(&r.connectedFlag, 1)
	r.mtx.Lock()
	for _, x := range r.lookupdHTTPAddrs {
		if x == addr {
			r.mtx.Unlock()
			return nil
		}
	}
	r.lookupdHTTPAddrs = append(r.lookupdHTTPAddrs, addr)
	numLookupd := len(r.lookupdHTTPAddrs)
	r.mtx.Unlock()
	if numLookupd == 1 {
		r.queryLookupd()
		r.wg.Add(1)
		go r.lookupdLoop()
	}
	return nil
}
func (r *Consumer) ConnectToNSQLookupds(addresses []string) error {
	for _, addr := range addresses {
		err := r.ConnectToNSQLookupd(addr)
		if err != nil {
			return err
		}
	}
	return nil
}
func validatedLookupAddr(addr string) error {
	if strings.Contains(addr, "/") {
		_, err := url.Parse(addr)
		if err != nil {
			return err
		}
		return nil
	}
	if !strings.Contains(addr, ":") {
		return errors.New("missing port")
	}
	return nil
}