// Consumes a partition
334047
func (cg *ConsumerGroup) partitionConsumer(topic string, partition int32, messages chan<- *sarama.ConsumerMessage, errors chan<- error, wg *sync.WaitGroup, stopper <-chan struct{}) {
	defer wg.Done()

	// Since ProcessingTimeout is the amount of time we'll wait for the final batch
	// of messages to be processed before releasing a partition, we need to wait slightly
	// longer than that before timing out here to ensure that another consumer has had
	// enough time to release the partition. Hence, +2 seconds.
	maxRetries := int(cg.config.Offsets.ProcessingTimeout/time.Second) + 2
partitionClaimLoop:
	for tries := 0; tries < maxRetries; tries++ {
		select {
		case <-stopper:
			return
		case <-time.After(1 * time.Second):
			if err := cg.instance.ClaimPartition(topic, partition); err == nil {
				break partitionClaimLoop
			} else if tries+1 < maxRetries {
				if err == kazoo.ErrPartitionClaimedByOther {
					// Another consumer still owns this partition. We should wait longer for it to release it.
				} else {
					// An unexpected error occurred. Log it and continue trying until we hit the timeout.
					cg.Logf("%s/%d :: FAILED to claim partition on attempt %v of %v; retrying in 1 second. Error: %v", topic, partition, tries+1, maxRetries, err)
				}
			} else {
				cg.Logf("%s/%d :: FAILED to claim the partition: %s\n", topic, partition, err)
				cg.errors <- &sarama.ConsumerError{
					Topic:     topic,
					Partition: partition,
					Err:       err,
				}
				return
			}
		}
	}

	defer func() {
		err := cg.instance.ReleasePartition(topic, partition)
		if err != nil {
			cg.Logf("%s/%d :: FAILED to release partition: %s\n", topic, partition, err)
			cg.errors <- &sarama.ConsumerError{
				Topic:     topic,
				Partition: partition,
				Err:       err,
			}
		}
	}()

	nextOffset, err := cg.offsetManager.InitializePartition(topic, partition)
	if err != nil {
		cg.Logf("%s/%d :: FAILED to determine initial offset: %s\n", topic, partition, err)
		return
	}

	if nextOffset >= 0 {
		cg.Logf("%s/%d :: Partition consumer starting at offset %d.\n", topic, partition, nextOffset)
	} else {
		nextOffset = cg.config.Offsets.Initial
		if nextOffset == sarama.OffsetOldest {
			cg.Logf("%s/%d :: Partition consumer starting at the oldest available offset.\n", topic, partition)
		} else if nextOffset == sarama.OffsetNewest {
			cg.Logf("%s/%d :: Partition consumer listening for new messages only.\n", topic, partition)
		}
	}

	consumer, err := cg.consumePartition(topic, partition, nextOffset)

	if err != nil {
		cg.Logf("%s/%d :: FAILED to start partition consumer: %s\n", topic, partition, err)
		return
	}

	defer consumer.Close()

	err = nil
	var lastOffset int64 = -1 // aka unknown
partitionConsumerLoop:
	for {
		select {
		case <-stopper:
			break partitionConsumerLoop

		case err := <-consumer.Errors():
			if err == nil {
				cg.Logf("%s/%d :: Consumer encountered an invalid state: re-establishing consumption of partition.\n", topic, partition)

				// Errors encountered (if any) are logged in the consumerPartition function
				var cErr error
				consumer, cErr = cg.consumePartition(topic, partition, lastOffset)
				if cErr != nil {
					break partitionConsumerLoop
				}
				continue partitionConsumerLoop
			}

			for {
				select {
				case errors <- err:
					continue partitionConsumerLoop

				case <-stopper:
					break partitionConsumerLoop
				}
			}

		case message := <-consumer.Messages():
			if message == nil {
				cg.Logf("%s/%d :: Consumer encountered an invalid state: re-establishing consumption of partition.\n", topic, partition)

				// Errors encountered (if any) are logged in the consumerPartition function
				var cErr error
				consumer, cErr = cg.consumePartition(topic, partition, lastOffset)
				if cErr != nil {
					break partitionConsumerLoop
				}
				continue partitionConsumerLoop

			}

			for {
				select {
				case <-stopper:
					break partitionConsumerLoop

				case messages <- message:
					lastOffset = message.Offset
					continue partitionConsumerLoop
				}
			}
		}
	}

	cg.Logf("%s/%d :: Stopping partition consumer at offset %d\n", topic, partition, lastOffset)
	if err := cg.offsetManager.FinalizePartition(topic, partition, lastOffset, cg.config.Offsets.ProcessingTimeout); err != nil {
		cg.Logf("%s/%d :: %s\n", topic, partition, err)
	}
}
1159518
func (w *Worker) DiskUsage(ctx context.Context, opt client.DiskUsageInfo) ([]*client.UsageInfo, error) {
	return w.CacheManager.DiskUsage(ctx, opt)
}
1208611
func (p *Parser) consumeWhitespace() {
	if tok, _, _ := p.Scan(); tok != WS {
		p.Unscan()
	}
}
746322
func (d *doubleStar) Consume(path []string, isDir bool) ([]string, bool) {
	if len(path) == 0 {
		return path, false
	}

	// If there are no remaining tokens to match, allow matching the entire
	// path.
	if d.Until == nil {
		return nil, true
	}

	for i := len(path); i > 0; i-- {
		rest, ok := d.Until.Consume(path[i:], false)
		if ok {
			return rest, ok
		}
	}

	// If no match has been found, we assume that the '**' token matches the
	// empty string, and defer pattern matching to the rest of the path.
	return d.Until.Consume(path, isDir)
}
234218
func (b *RouteBuilder) Consumes(mimeTypes ...string) *RouteBuilder {
	b.consumes = mimeTypes
	return b
}
530661
func Consume(shardID string, streamName string, region string, shardIteratorType string, eventChannel chan []byte) {
	svc := kinesis.New(session.New(), &aws.Config{Region: aws.String(region)})
	getShardIteratorParams := &kinesis.GetShardIteratorInput{
		ShardId:           aws.String(shardID),           // Required
		ShardIteratorType: aws.String(shardIteratorType), // Required
		StreamName:        aws.String(streamName),        // Required
	}
	shardIterator, err := svc.GetShardIterator(getShardIteratorParams)
	if err != nil {
		log.Panic(err)
	}

	nextShardIterator := shardIterator.ShardIterator
	for {
		getRecordsParams := &kinesis.GetRecordsInput{
			ShardIterator: nextShardIterator, // Required
			Limit:         aws.Int64(1000),
		}
		getRecordsResp, err := svc.GetRecords(getRecordsParams)
		if err != nil {
			log.Panic(err)
		}
		for _, record := range getRecordsResp.Records {
			eventChannel <- record.Data
		}
		nextShardIterator = getRecordsResp.NextShardIterator
		time.Sleep(time.Second)
	}
}
771164
func (cons *Socket) Consume(workers *sync.WaitGroup) {
	cons.AddMainWorker(workers)
	defer cons.closeListener()

	if cons.protocol == "udp" {
		go tgo.WithRecoverShutdown(cons.listenUDP)
	} else {
		go tgo.WithRecoverShutdown(cons.listen)
	}

	cons.ControlLoop()
}
1297099
func (p *Partition) Compact() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.compact()
}
1080918
func (l *Logger) consume() {
	go func() {
		// Process the single next task in sequence until completion,
		// then consume the next task.
		for task := range l.tasks {
			l.logTask(task)
		}
	}()

	defer close(l.tasks)

	pending := make([]Task, 0)

	for {
		// If there is a pending task, "peek" it off of the set of
		// pending tasks.
		var next Task
		if len(pending) > 0 {
			next = pending[0]
		}

		if next == nil {
			// If there was no pending task, wait for either a)
			// l.queue to close, or b) a new task to be submitted.
			task, ok := <-l.queue
			if !ok {
				// If the queue is closed, no more new tasks may
				// be added.
				return
			}

			// Otherwise, add a new task to the set of tasks to
			// process immediately, since there is no current
			// buffer.
			l.tasks <- task
		} else {
			// If there is a pending task, wait for either a) a
			// write to process the task to become non-blocking, or
			// b) a new task to enter the queue.
			select {
			case task, ok := <-l.queue:
				if !ok {
					// If the queue is closed, no more tasks
					// may be added.
					return
				}
				// Otherwise, add the next task to the set of
				// pending, active tasks.
				pending = append(pending, task)
			case l.tasks <- next:
				// Or "pop" the peeked task off of the pending
				// set.
				pending = pending[1:]
			}
		}
	}
}
882346
func (d *MutexDatastore) DiskUsage() (uint64, error) {
	d.RLock()
	defer d.RUnlock()
	return ds.DiskUsage(d.child)
}
1255707
func consumeKeyword(l *lexer, t TokenType) {
	for {
		if r := l.next(); !unicode.IsLetter(r) || r == eof {
			l.backup()
			l.emit(t)
			break
		}
	}
}
1005068
func ConsumeAll(lexer Lexer) ([]Token, error) {
	tokens := []Token{}
	for {
		token, err := lexer.Next()
		if err != nil {
			return nil, err
		}
		tokens = append(tokens, token)
		if token.Type == EOF {
			return tokens, nil
		}
	}
}
129049
func (cons *Kafka) Consume(workers *sync.WaitGroup) {
	cons.SetWorkerWaitGroup(workers)

	if err := cons.startAllConsumers(); err != nil {
		cons.Logger.WithError(err).Error("Kafka client error")
		time.AfterFunc(cons.config.Net.DialTimeout, func() { cons.Consume(workers) })
		return
	}

	defer func() {
		cons.client.Close()
		cons.dumpIndex()
	}()

	cons.TickerControlLoop(cons.persistTimeout, cons.dumpIndex)
}
311094
func (l *Logs) consume(name string, ch chan *Event, done chan error) {
	log := Log{
		Config:    l.Config,
		GroupName: name,
		Log:       log.WithField("group", name),
	}

	for event := range log.Start() {
		ch <- event
	}

	done <- log.Err()
}
381822
func (pc *KafkaPartitionConsumer) Stop() {
	log.Info("Stopping partition consumer for topic %s, partition %d", pc.topic, pc.partition)
	pc.stop <- struct{}{}
	pc.metrics.Stop()
}
607604
func (cl *CommandList) Consume(consume bool) {
	if consume {
		cl.cmdQ.PushBack(&command{"consume 1", nil, cmdNoReturn})
	} else {
		cl.cmdQ.PushBack(&command{"consume 0", nil, cmdNoReturn})
	}
}
1169453
func (c *KafkaConsumer) PartitionConsumerMetrics(topic string, partition int32) (PartitionConsumerMetrics, error) {
	if !c.config.EnableMetrics {
		return nil, ErrMetricsDisabled
	}

	c.partitionConsumersLock.Lock()
	defer c.partitionConsumersLock.Unlock()

	if !c.exists(topic, partition) {
		log.Info("Partition consumer for topic %s, partition %d does not exist", topic, partition)
		return nil, ErrPartitionConsumerDoesNotExist
	}

	return c.partitionConsumers[topic][partition].Metrics()
}