package sink

import (
	"encoding/json"
	"expvar"
	"sync/atomic"

	"code.justin.tv/common/golibs/pkgpath"
	"code.justin.tv/ear/kinesis/combiner"
)

// Generic map for storing stats on sinks
var expStats *expvar.Map

func init() {
	pkg, _ := pkgpath.Caller(0)
	expStats = expvar.NewMap(pkg)
}

// Statistics on a sink
type sinkStat struct {
	// Number of Blobs accepted by the sink
	accepted *int64

	// Number of Blobs skipped by the sink
	skipped *int64

	// Input channel for the sink (not exposed - just held onto for
	// stats)
	ch chan *combiner.Blob
}

func newSinkStat(ch chan *combiner.Blob) *sinkStat {
	return &sinkStat{
		accepted: new(int64),
		skipped:  new(int64),
		ch:       ch,
	}
}

func (s *sinkStat) RecordAccepted() {
	atomic.AddInt64(s.accepted, 1)
}

func (s *sinkStat) RecordSkipped() {
	atomic.AddInt64(s.skipped, 1)
}

func (s *sinkStat) String() string {

	type marshaled struct {
		CurrentBufferLen int
		CurrentBufferCap int
		Accepted         int64
		Skipped          int64
	}

	m := &marshaled{
		CurrentBufferLen: len(s.ch),
		CurrentBufferCap: cap(s.ch),
		Accepted:         atomic.LoadInt64(s.accepted),
		Skipped:          atomic.LoadInt64(s.skipped),
	}

	v, _ := json.Marshal(m)
	return string(v)
}
