package scanproto

import (
	"sync"
	"time"

	"code.justin.tv/release/trace/api"
)

type limitedTxSrc struct {
	base  TransactionSource
	limit time.Duration

	wg sync.WaitGroup

	timer *time.Timer
}

func (lts *limitedTxSrc) Run() {
	lts.wg.Add(1)
	lts.timer = time.AfterFunc(lts.limit, lts.Stop)
	go func() {
		defer lts.wg.Done()
		lts.base.Run()
	}()
	lts.wg.Wait()
}

func (lts *limitedTxSrc) Transactions() <-chan *api.Transaction {
	return lts.base.Transactions()
}

func (lts *limitedTxSrc) Errors() <-chan error {
	return lts.base.Errors()
}

func (lts *limitedTxSrc) Stop() {
	lts.base.Stop()
	lts.wg.Wait()
}

// Limit a TransactionSource to only run for d duration. It will call
// the base TransactionSource's Stop function after d time passes, which
// might take time, so it's not guaranteed to finish in less than d time.
func LimitedTransactionSource(s TransactionSource, d time.Duration) TransactionSource {
	return &limitedTxSrc{
		base:  s,
		limit: d,
	}
}
