package sink

import (
	"math/rand"
	"sync"

	"code.justin.tv/ear/kinesis/combiner"
)

// Sample only passes a fraction of data into a sink. It samples
// uniformly, without using the key of the data blobs.
func Sample(rate float64, base Sink) Sink {
	s := &samplerSink{
		rate: rate,
		sink: base,
	}

	return s.run
}

type samplerSink struct {
	rate    float64
	sampled chan *combiner.Blob

	sink Sink
}

func (s *samplerSink) run(data <-chan *combiner.Blob) {
	s.sampled = make(chan *combiner.Blob, cap(data))

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		s.sink(s.sampled)
		wg.Done()
	}()

	for b := range data {
		if rand.Float64() < s.rate {
			s.sampled <- b
		}
	}
	close(s.sampled)
	wg.Wait()
}
