dom/telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Improve the HTTP JSON plugin README with more examples.

Browse Source
This commit is contained in:
Cameron Sparr
2015-11-02 16:19:37 -07:00
parent 422d240afb
commit befc906167
3 changed files with 107 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
plugins/kafka_consumer/kafka_consumer.go
View File

@@ -74,7 +74,11 @@ func (k *Kafka) Gather(acc plugins.Accumulator) error {
k.Consumer.Close()
}()
go readFromKafka(k.Consumer.Messages(), metricQueue, k.BatchSize, k.Consumer.CommitUpto, halt)
go readFromKafka(k.Consumer.Messages(),
metricQueue,
k.BatchSize,
k.Consumer.CommitUpto,
halt)
}
return emitMetrics(k, acc, metricQueue)
@@ -105,7 +109,13 @@ const millisecond = 1000000 * time.Nanosecond
type ack func(*sarama.ConsumerMessage) error
func readFromKafka(kafkaMsgs <-chan *sarama.ConsumerMessage, metricProducer chan<- []byte, maxBatchSize int, ackMsg ack, halt <-chan bool) {
func readFromKafka(
kafkaMsgs <-chan *sarama.ConsumerMessage,
metricProducer chan<- []byte,
maxBatchSize int,
ackMsg ack,
halt <-chan bool,
) {
batch := make([]byte, 0)
currentBatchSize := 0
timeout := time.After(500 * millisecond)