Prez - Replay aggregator

aggregator.go

func (a *replayAggregator) Run() {
	var timer = time.NewTimer(a.flushFrequency)
	var elapsed <-chan time.Time
	var out chan []*sarama.ConsumerMessage

	for {
		select {
		case msg := <-a.in:
			if !a.slicer.add(msg) {
				a.out <- a.slicer.flush()
				a.slicer.reset()
				elapsed = nil
			}
			if elapsed == nil {
				timer.Reset(a.flushFrequency)
				elapsed = timer.C
			}
		case <-elapsed:
			out = a.out
			elapsed = nil
		case out <- a.slicer.flush():
			a.slicer.reset()
			out = nil
		case err := <-a.errs:
			log.Printf("consuming error: %s", err)
		case <-a.closing:
			log.Println("closing replay aggregator")
			timer.Stop()
			return
		}
	}
}