func (prod *StatsdMetrics) Produce(workers *sync.WaitGroup) {
	prod.AddMainWorker(workers)
	prod.TickerMessageControlLoop(prod.bufferMessage, prod.flushFrequency, prod.sendBatchOnTimeOut)
}
Produce writes to stdout or stderr.
package producer
import (
	"strconv"
	"sync"
	"time"
	"gollum/core"
	"github.com/quipo/statsd"
)
type StatsdMetrics struct {
	core.BufferedProducer `gollumdoc:"embed_type"`
	streamMap             map[core.MessageStreamID]string
	client                *statsd.StatsdClient
	batch                 core.MessageBatch
	flushFrequency        time.Duration `config:"Batch/TimeoutSec" default:"10" metric:"sec"`
	useMessage            bool          `config:"UseMessage" default:"false"`
	useGauge              bool          `config:"UseGauge" default:"false"`
}
func init() {
	core.TypeRegistry.Register(StatsdMetrics{})
}
func (prod *StatsdMetrics) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)
	prod.streamMap = conf.GetStreamMap("StreamMapping", "")
	prod.batch = core.NewMessageBatch(int(conf.GetInt("Batch/MaxMessages", 500)))
	server := conf.GetString("Server", "localhost:8125")
	prefix := conf.GetString("Prefix", "gollum.")
	prod.client = statsd.NewStatsdClient(server, prefix)
	conf.Errors.Push(prod.client.CreateSocket())
}
func (prod *StatsdMetrics) bufferMessage(msg *core.Message) {
	prod.batch.AppendOrFlush(msg, prod.sendBatch, prod.IsActiveOrStopping, prod.TryFallback)
}
func (prod *StatsdMetrics) sendBatchOnTimeOut() {
	if prod.batch.ReachedTimeThreshold(prod.flushFrequency) || prod.batch.ReachedSizeThreshold(prod.batch.Len()/2) {
		prod.sendBatch()
	} else if prod.useGauge && prod.batch.IsEmpty() {
		prod.transformMessages(make([]*core.Message, 0))
	}
}
func (prod *StatsdMetrics) sendBatch() {
	prod.batch.Flush(prod.transformMessages)
}
func (prod *StatsdMetrics) transformMessages(messages []*core.Message) {
	metricValues := make(map[string]int64)
	for _, msg := range messages {
		metricName, streamMapped := prod.streamMap[msg.GetStreamID()]
		if !streamMapped {
			metricName, streamMapped = prod.streamMap[core.WildcardStreamID]
			if !streamMapped {
				metricName = core.StreamRegistry.GetStreamName(msg.GetStreamID())
				prod.streamMap[msg.GetStreamID()] = metricName
			}
		}
		_, metricMapped := metricValues[metricName]
		if !metricMapped {
			metricValues[metricName] = int64(0)
		}
		if prod.useMessage {
			if val, err := strconv.ParseInt(msg.String(), 10, 64); err == nil {
				metricValues[metricName] += val
			} else {
				prod.Logger.Warning("message was skipped")
			}
		} else {
			metricValues[metricName] += int64(1)
		}
	}
	if prod.useGauge {
		for _, metricName := range prod.streamMap {
			if _, metricMapped := metricValues[metricName]; !metricMapped {
				metricValues[metricName] = int64(0)
			}
		}
	}
	for metric, val := range metricValues {
		if prod.useGauge {
			prod.client.Gauge(metric, val)
		} else {
			prod.client.Incr(metric, val)
		}
	}
}
func (prod *StatsdMetrics) close() {
	defer prod.WorkerDone()
	prod.CloseMessageChannel(prod.bufferMessage)
	prod.batch.Close(prod.transformMessages, prod.GetShutdownTimeout())
	prod.client.Close()
}