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
}
ConnectToNSQLookupd adds an nsqlookupd address to the list for this Consumer instance.

If it is the first to be added, it initiates an HTTP request to discover nsqd
producers for the configured topic.

A goroutine is spawned to handle continual polling.
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)
	}
}