package serf

import (
	"time"
)

// coalescer is a simple interface that must be implemented to be
// used inside of a coalesceLoop
type coalescer interface {
	// Can the coalescer handle this event, if not it is
	// directly passed through to the destination channel
	Handle(Event) bool

	// Invoked to coalesce the given event
	Coalesce(Event)

	// Invoked to flush the coalesced events
	Flush(outChan chan<- Event)
}

// coalescedEventCh returns an event channel where the events are coalesced
// using the given coalescer.
func coalescedEventCh(outCh chan<- Event, shutdownCh <-chan struct{},
	cPeriod time.Duration, qPeriod time.Duration, c coalescer) chan<- Event {
	inCh := make(chan Event, 1024)
	go coalesceLoop(inCh, outCh, shutdownCh, cPeriod, qPeriod, c)
	return inCh
}

// coalesceLoop is a simple long-running routine that manages the high-level
// flow of coalescing based on quiescence and a maximum quantum period.
func coalesceLoop(inCh <-chan Event, outCh chan<- Event, shutdownCh <-chan struct{},
	coalescePeriod time.Duration, quiescentPeriod time.Duration, c coalescer) {
	var quiescent <-chan time.Time
	var quantum <-chan time.Time
	shutdown := false

INGEST:
	// Reset the timers
	quantum = nil
	quiescent = nil

	for {
		select {
		case e := <-inCh:
			// Ignore any non handled events
			if !c.Handle(e) {
				outCh <- e
				continue
			}

			// Start a new quantum if we need to
			// and restart the quiescent timer
			if quantum == nil {
				quantum = time.After(coalescePeriod)
			}
			quiescent = time.After(quiescentPeriod)

			// Coalesce the event
			c.Coalesce(e)

		case <-quantum:
			goto FLUSH
		case <-quiescent:
			goto FLUSH
		case <-shutdownCh:
			shutdown = true
			goto FLUSH
		}
	}

FLUSH:
	// Flush the coalesced events
	c.Flush(outCh)

	// Restart ingestion if we are not done
	if !shutdown {
		goto INGEST
	}
}