package main

import (
	"fmt"
	"sync"

	"github.com/cactus/go-statsd-client/statsd"
)

// number of Metrics to hold in write buffers
var writeBuffer = 1024

// metricSink is an interface to represent a destination for
// computed Metrics.
type MetricSink interface {
	Consume(<-chan *Metric)
}

type statsdMetricSink struct {
	stats   statsd.Statter
	workers int
}

func NewStatsdMetricSink(hostport string, workers int) (MetricSink, error) {
	stats, err := statsd.NewBufferedClient(hostport, traceStatsdPrefix, statsdFlushInterval, statsdFlushSize)
	if err != nil {
		return nil, err
	}

	sink := statsdMetricSink{
		stats:   stats,
		workers: workers,
	}

	return &sink, nil
}

func (ms *statsdMetricSink) Consume(metrics <-chan *Metric) {
	var wg sync.WaitGroup

	for i := 0; i < ms.workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for m := range metrics {
				ms.stats.Inc(fmt.Sprintf("v1.%s.%s.%s.%s.%d", m.service, m.environment, m.hostname, m.route, m.status), 1, statsdRate)
				ms.stats.TimingDuration(fmt.Sprintf("v1.%s.%s.%s.%s.duration", m.service, m.environment, m.hostname, m.route), m.duration, statsdRate)
			}
		}()
	}

	wg.Wait()
}
