func (c *Consumer) getNumActiveClaims() (ct int) {
	c.lock.RLock()
	defer c.lock.RUnlock()
	for _, topicClaims := range c.claims {
		for _, cl := range topicClaims {
			if !cl.Terminated() {
				ct++
			}
		}
	}
	return
}
getNumActiveClaims returns the number of claims actively owned by this Consumer.
package marshal
import (
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"github.com/zorkian/kafka/proto"
)
type CommitToken struct {
	topic  string
	partID int
	offset int64
}
type Message proto.Message
func (m *Message) CommitToken() CommitToken {
	return CommitToken{
		topic:  m.Topic,
		partID: int(m.Partition),
		offset: m.Offset,
	}
}
type ConsumerOptions struct {
	FastReclaim bool
	ClaimEntireTopic bool
	GreedyClaims bool
	ReleaseClaimsIfBehind bool
	MaximumClaims int
}
type Consumer struct {
	alive    *int32
	marshal  *Marshaler
	topics   []string
	options  ConsumerOptions
	messages chan *Message
	topicClaimsChan    chan map[string]bool
	topicClaimsUpdated chan struct{}
	lock       *sync.RWMutex
	rand       *rand.Rand
	partitions map[string]int
	claims     map[string]map[int]*claim
	stopChan   chan struct{}
	doneChan   chan struct{}
}
func (m *Marshaler) NewConsumer(topicNames []string, options ConsumerOptions) (*Consumer, error) {
	if m.Terminated() {
		return nil, errors.New("Marshaler has terminated, no new consumers can be created")
	}
	if len(topicNames) > 1 && !options.ClaimEntireTopic {
		return nil, errors.New("ClaimEntireTopic must be set if provided more than one topic")
	} else if len(topicNames) == 0 {
		return nil, errors.New("must provide at least one topic")
	}
	partitions := make(map[string]int)
	for _, topic := range topicNames {
		partitions[topic] = m.Partitions(topic)
	}
	c := &Consumer{
		alive:              new(int32),
		marshal:            m,
		topics:             topicNames,
		partitions:         partitions,
		options:            options,
		messages:           make(chan *Message, m.cluster.options.MaxMessageQueue),
		lock:               &sync.RWMutex{},
		rand:               rand.New(rand.NewSource(time.Now().UnixNano())),
		claims:             make(map[string]map[int]*claim),
		topicClaimsChan:    make(chan map[string]bool),
		topicClaimsUpdated: make(chan struct{}, 1),
		stopChan:           make(chan struct{}),
		doneChan:           make(chan struct{}),
	}
	atomic.StoreInt32(c.alive, 1)
	m.addNewConsumer(c)
	c.lock.Lock()
	defer c.lock.Unlock()
	go c.sendTopicClaimsLoop()
	if c.options.FastReclaim {
		claimedTopics := make(map[string]bool)
		for topic, partitionCount := range c.partitions {
			for partID := 0; partID < partitionCount; partID++ {
				cl := c.marshal.GetPartitionClaim(topic, partID)
				if !cl.Claimed() ||
					cl.ClientID != c.marshal.ClientID() ||
					cl.GroupID != c.marshal.GroupID() {
					continue
				}
				log.Infof("[%s:%d] attempting to fast-reclaim", topic, partID)
				if _, ok := c.claims[topic]; !ok {
					c.claims[topic] = make(map[int]*claim)
				}
				if options.ClaimEntireTopic {
					if partID == 0 {
						claimedTopics[topic] = true
					}
					if !claimedTopics[topic] {
						log.Infof("[%s:%d] blocked fast-reclaim because topic is not claimed",
							topic, partID)
						continue
					}
				}
				claim := newClaim(
					topic, partID, c.marshal, c, c.messages, options)
				if claim == nil {
					log.Warningf("[%s:%d] failed to fast-reclaim", topic, partID)
				} else {
					c.claims[topic][partID] = claim
					go claim.healthCheckLoop()
					c.sendTopicClaimsUpdate()
				}
			}
			if options.ClaimEntireTopic && len(claimedTopics) >= options.MaximumClaims {
				log.Infof("reached max-topics for fast-reclaim. Claimed topics: %v",
					claimedTopics)
				break
			}
		}
	}
	go c.manageClaims()
	return c, nil
}
func NewConsumerOptions() ConsumerOptions {
	return ConsumerOptions{
		FastReclaim:           true,
		ClaimEntireTopic:      false,
		GreedyClaims:          false,
		ReleaseClaimsIfBehind: true,
	}
}
func (c *Consumer) defaultTopic() string {
	c.lock.RLock()
	defer c.lock.RUnlock()
	if len(c.partitions) > 1 {
		log.Errorf("attempted to claim partitions for more than one topic")
		go c.Terminate(false)
		return ""
	}
	for topic := range c.partitions {
		return topic
	}
	log.Errorf("couldn't find default topic!")
	go c.Terminate(false)
	return ""
}
func (c *Consumer) defaultTopicPartitions() int {
	c.lock.RLock()
	defer c.lock.RUnlock()
	if len(c.partitions) > 1 {
		log.Errorf("attempted to claim partitions for more than one topic")
		go c.Terminate(false)
		return 0
	}
	for _, partitions := range c.partitions {
		return partitions
	}
	log.Errorf("couldn't find default topic!")
	go c.Terminate(false)
	return 0
}
func (c *Consumer) claimTerminated(cl *claim, released bool) {
	if !c.options.ClaimEntireTopic {
		return
	}
	defer c.sendTopicClaimsUpdate()
	c.lock.RLock()
	defer c.lock.RUnlock()
	for _, claim := range c.claims[cl.topic] {
		if cl != claim {
			if released {
				go claim.Release()
			} else {
				go claim.Terminate()
			}
		}
	}
}
func (c *Consumer) tryClaimPartition(topic string, partID int) bool {
	if c.options.ClaimEntireTopic {
		if c.isTopicClaimLimitReached(topic) {
			return false
		}
	} else {
		if c.isClaimLimitReached() {
			return false
		}
	}
	currentClaim := c.marshal.GetPartitionClaim(topic, partID)
	if currentClaim.Claimed() {
		return false
	}
	newClaim := newClaim(topic, partID, c.marshal, c, c.messages, c.options)
	if newClaim == nil {
		return false
	}
	go newClaim.healthCheckLoop()
	c.lock.Lock()
	defer c.lock.Unlock()
	if newClaim.Terminated() {
		return false
	}
	if c.Terminated() {
		go newClaim.Release()
		return false
	}
	topicClaims, ok := c.claims[topic]
	if ok {
		oldClaim, ok := topicClaims[partID]
		if ok && oldClaim != nil {
			if !oldClaim.Terminated() {
				log.Errorf("Internal double-claim for %s:%d.", topic, partID)
				log.Errorf("This is a catastrophic error. We're terminating Marshal.")
				log.Errorf("No further messages will be available. Please restart.")
				go newClaim.Release()
				go c.terminateAndCleanup(false, false)
				go func() {
					c.marshal.PrintState()
					c.marshal.Terminate()
				}()
				return false
			}
		}
	}
	if _, ok := c.claims[topic]; !ok {
		c.claims[topic] = make(map[int]*claim)
	}
	c.claims[topic][partID] = newClaim
	return true
}
func (c *Consumer) rndIntn(n int) int {
	c.lock.Lock()
	defer c.lock.Unlock()
	return c.rand.Intn(n)
}
func (c *Consumer) releaseClaims() {
	c.lock.Lock()
	defer c.lock.Unlock()
	for topic, partitions := range c.claims {
		for partID, claim := range partitions {
			if !claim.Terminated() {
				log.Warningf("[%s:%d] Consumer still paused, releasing claim",
					topic, partID)
				claim.Release()
			}
		}
		c.claims[topic] = make(map[int]*claim)
	}
}
func (c *Consumer) claimPartitions() {
	func() {
		c.lock.RLock()
		defer c.lock.RUnlock()
		if len(c.partitions) > 1 {
			log.Errorf("attempted to claim partitions for more than a single topic")
			go c.Terminate(false)
		}
	}()
	topic := c.defaultTopic()
	partitions := c.defaultTopicPartitions()
	if partitions <= 0 {
		return
	}
	if c.isClaimLimitReached() {
		return
	}
	offset := c.rndIntn(partitions)
	for i := 0; i < partitions; i++ {
		partID := (i + offset) % partitions
		lastClaim := c.marshal.GetLastPartitionClaim(topic, partID)
		if lastClaim.Claimed() {
			continue
		}
		if lastClaim.GroupID == c.marshal.groupID &&
			lastClaim.ClientID == c.marshal.clientID {
			if time.Now().Unix()-lastClaim.LastRelease < HeartbeatInterval {
				log.Infof("[%s:%d] skipping unclaimed partition because we recently released it",
					topic, partID)
				continue
			} else {
				log.Infof("[%s:%d] reclaiming because we released it a while ago",
					topic, partID)
			}
		}
		if !c.tryClaimPartition(topic, partID) {
			continue
		}
		if !c.options.GreedyClaims {
			break
		}
	}
}
func (c *Consumer) isTopicClaimLimitReached(topic string) bool {
	if c.options.MaximumClaims <= 0 {
		return false
	}
	c.lock.RLock()
	defer c.lock.RUnlock()
	claimed := make(map[string]bool)
	for topic, topicClaims := range c.claims {
		if claim, ok := topicClaims[0]; ok && !claim.Terminated() {
			claimed[topic] = true
		}
	}
	if !claimed[topic] && len(claimed) >= c.options.MaximumClaims {
		return true
	}
	return false
}
func (c *Consumer) claimTopics() {
	defer c.sendTopicClaimsUpdate()
	topicPartitions := make(map[string]int)
	func() {
		c.lock.RLock()
		defer c.lock.RUnlock()
		for k, v := range c.partitions {
			topicPartitions[k] = v
		}
	}()
	for topic, partitions := range topicPartitions {
		if partitions <= 0 {
			continue
		}
		lastClaim := c.marshal.GetLastPartitionClaim(topic, 0)
		if lastClaim.Claimed() {
			if lastClaim.GroupID != c.marshal.groupID ||
				lastClaim.ClientID != c.marshal.clientID {
				continue
			}
		} else {
			log.Infof("[%s] attempting to claim topic (key partition 0)", topic)
			if c.isTopicClaimLimitReached(topic) {
				log.Debugf("[%s] blocked claiming topic due to limit: %d",
					topic, c.options.MaximumClaims)
				continue
			}
			if !c.tryClaimPartition(topic, 0) {
				continue
			}
			log.Infof("[%s] claimed topic (key partition 0) successfully", topic)
			c.sendTopicClaimsUpdate()
		}
		for partID := 1; partID < partitions; partID++ {
			if !c.marshal.Claimed(topic, partID) {
				log.Infof("[%s:%d] claiming partition (topic claim mode)", topic, partID)
				c.tryClaimPartition(topic, partID)
			}
		}
	}
}
func (c *Consumer) sendTopicClaimsUpdate() {
	select {
	case c.topicClaimsUpdated <- struct{}{}:
	default:
	}
}
func (c *Consumer) sendTopicClaimsLoop() {
	defer close(c.topicClaimsChan)
	lastClaims := make(map[string]bool)
	keepRunning := true
	for keepRunning {
		select {
		case <-c.topicClaimsUpdated:
		case <-c.stopChan:
			keepRunning = false
		}
		claims, err := c.GetCurrentTopicClaims()
		if err != nil {
			log.Errorf("Failed to send topic claims update: %s", err)
			continue
		}
		anyUpdates := false
		for topic, claimed := range lastClaims {
			stillClaimed, ok := claims[topic]
			if !ok {
				anyUpdates = true
			} else if claimed != stillClaimed {
				anyUpdates = true
			}
		}
		for topic := range claims {
			if _, ok := lastClaims[topic]; !ok {
				anyUpdates = true
			}
		}
		lastClaims = claims
		if !anyUpdates {
			continue
		}
		select {
		case <-c.topicClaimsChan:
		default:
		}
		c.topicClaimsChan <- claims
	}
}
func (c *Consumer) updatePartitionCounts() {
	c.lock.Lock()
	defer c.lock.Unlock()
	for _, topic := range c.marshal.Topics() {
		if _, ok := c.partitions[topic]; ok {
			c.partitions[topic] = c.marshal.Partitions(topic)
		}
	}
}
func (c *Consumer) manageClaims() {
	for !c.Terminated() {
		c.updatePartitionCounts()
		if c.marshal.cluster.IsGroupPaused(c.marshal.GroupID()) {
			c.releaseClaims()
		} else {
			if c.options.ClaimEntireTopic {
				c.claimTopics()
			} else {
				c.claimPartitions()
			}
		}
		time.Sleep(time.Duration(c.rndIntn(3000)) * time.Millisecond)
	}
}
func (c *Consumer) Terminated() bool {
	return atomic.LoadInt32(c.alive) == 0
}
func (c *Consumer) terminateAndCleanup(release bool, remove bool) bool {
	if !atomic.CompareAndSwapInt32(c.alive, 1, 0) {
		return false
	}
	close(c.stopChan)
	defer close(c.doneChan)
	latestTopicClaims := make(map[string]bool)
	releasedTopics := make(map[string]bool)
	c.lock.Lock()
	defer c.lock.Unlock()
	for topic, topicClaims := range c.claims {
		for partID, claim := range topicClaims {
			if claim != nil {
				if release {
					claim.Release()
					if partID == 0 {
						releasedTopics[topic] = true
					}
				} else {
					claim.Terminate()
				}
			}
		}
	}
	close(c.messages)
	for topic := range c.claims {
		if !releasedTopics[topic] {
			latestTopicClaims[topic] = true
		}
	}
	if remove {
		c.marshal.removeConsumer(c)
	}
	c.sendTopicClaimsUpdate()
	return true
}
func (c *Consumer) Terminate(release bool) bool {
	return c.terminateAndCleanup(release, true)
}
func (c *Consumer) GetCurrentTopicClaims() (map[string]bool, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()
	if !c.options.ClaimEntireTopic {
		return nil, errors.New(
			"GetCurrentTopicClaims requires options.ClaimEntireTopic be set")
	}
	claimedTopics := make(map[string]bool)
	if c.Terminated() {
		return claimedTopics, nil
	}
	for topic := range c.partitions {
		cl := c.marshal.GetPartitionClaim(topic, 0)
		if cl.ClientID == c.marshal.ClientID() &&
			cl.GroupID == c.marshal.GroupID() {
			claimedTopics[topic] = true
		}
	}
	return claimedTopics, nil
}
func (c *Consumer) TopicClaims() <-chan map[string]bool {
	if !c.options.ClaimEntireTopic {
		err := fmt.Errorf(
			"GetCurrentTopicClaims is only relevant when ClaimEntireTopic is set")
		log.Error(err.Error())
	}
	return c.topicClaimsChan
}
func (c *Consumer) GetCurrentLag() int64 {
	c.lock.RLock()
	defer c.lock.RUnlock()
	var lag int64
	for _, topicClaims := range c.claims {
		for _, cl := range topicClaims {
			if !cl.Terminated() {
				lag += cl.GetCurrentLag()
			}
		}
	}
	return lag
}
func (c *Consumer) GetCurrentLoad() int {
	return c.getNumActiveClaims()
}