package sink

import (
	"expvar"
	"fmt"
	"reflect"
	"runtime"
	"sync"
	"time"

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

// broadcastReceiver represents a connection from a broadcaster to a
// single child sink
type broadcastReceiver struct {
	sink  Sink
	input chan *combiner.Blob
	stats *sinkStat
}

// how long a receiver has to accept a record before it will be skipped
const broadcastReceiveTimeout = time.Millisecond

// run starts the broadcastReceiver, and alerts wg when it finishes.
func (r *broadcastReceiver) run(wg *sync.WaitGroup) {
	r.sink(r.input)
	wg.Done()
}

// broadcaster sends input blobs to many sinks
type broadcaster struct {
	input  <-chan *combiner.Blob
	recvrs []*broadcastReceiver

	wg sync.WaitGroup
}

// newBroadcaster sets up a broadcaster which will route data to
// sinks, and adds an entry to the shared sink expvar stats map
func newBroadcaster(data <-chan *combiner.Blob, sinks ...Sink) *broadcaster {
	b := &broadcaster{
		input:  data,
		recvrs: make([]*broadcastReceiver, len(sinks)),
	}

	// Namespace stats under 'Broadcast'
	broadcastStats := new(expvar.Map).Init()
	expStats.Set("Broadcast", broadcastStats)

	for i, sink := range sinks {
		// Make a channel just for this sink, so each one gets a
		// different buffer.
		ch := make(chan *combiner.Blob, cap(data))

		// Store stats for each sink. Name them based on the sink's
		// function name and its index. We need the index to
		// deduplicate if the same type of sink is added multiple
		// times.
		stats := newSinkStat(ch)
		name := fmt.Sprintf("%s-%d", getFunctionName(sink), i)
		broadcastStats.Set(name, stats)

		recvr := &broadcastReceiver{
			sink:  sink,
			input: ch,
			stats: stats,
		}
		b.recvrs[i] = recvr

		// Immediately start the sinks. They won't have any data yet, though.
		b.wg.Add(1)
		go recvr.run(&b.wg)
	}
	return b
}

// broadcast sends data from the input to all child sinks, manages
// their stats, and blocks until all child sinks finish
// processing. Children receive data as non-blocking sends, so if
// their buffer fills, they will drop data.
func (b *broadcaster) broadcast() {
	for blob := range b.input {
		for _, recvr := range b.recvrs {
			// Do a nonblocking send to the receiver. If it fails, set
			// a timer; doing this under the default case avoids
			// starting a bajillion timers except when buffers are full.
			select {
			case recvr.input <- blob:
				// The happy case: The buffer isn't full.
				recvr.stats.RecordAccepted()
			default:
				// Unhappy case: buffer is full. Set a timeout to give
				// the receiver a chance to catch its breath.
				t := time.NewTimer(broadcastReceiveTimeout)
				select {
				case recvr.input <- blob:
					recvr.stats.RecordAccepted()
				case <-t.C:
					recvr.stats.RecordSkipped()
				}
				t.Stop()
			}
		}
	}
	for _, recvr := range b.recvrs {
		close(recvr.input)
	}
	b.wg.Wait()
}

// Broadcast sends blobs from the data channel to all sinks. It blocks
// until all sinks have consumed all data.
//
// Broadcasting uses non-blocking sends to write to each sink. If a
// sink's buffer fills up, it will skip sending blobs to the sink. The
// number of skipped blobs (as well as the number of accepted blobs)
// are recorded in an expvar. The expvar is stored under the key
// 'Broadcast'; if data is already stored under this key, then it will
// be overwritten when the Broadcast function is called, so you only
// get stats for one Broadcast at a time.
func Broadcast(data <-chan *combiner.Blob, sinks ...Sink) {
	b := newBroadcaster(data, sinks...)
	b.broadcast()
}

// Get the name of a function.
func getFunctionName(i interface{}) string {
	v := reflect.ValueOf(i)
	if v.Kind() != reflect.Func {
		return "unknown"
	}
	f := runtime.FuncForPC(v.Pointer())
	if f == nil {
		return "unknown"
	}
	return f.Name()
}
