Accumulator.go

type Accumulation struct {
	messages []interface{}
	t        *time.Timer
}

// Accumulator debounces and accumulates values then sends them to function passed in the Add method
// Implements a concurrency safe map that can accumulate messages based on a key
type Accumulator struct {
	acc   map[string]*Accumulation
	mu    sync.Mutex
	delay time.Duration
}

func MakeAccumulator(delay time.Duration) *Accumulator {
	return &Accumulator{
		acc:   make(map[string]*Accumulation),
		delay: delay,
	}
}

// Add creates a key in the map if it doesn't exist then assigns a callback value,
// then a waitTimer is called with that registeredCallback
// Subsequent calls to initialised entry in map doesn't re-register a callback function
// only the first one is used
func (d *Accumulator) Add(key string, message interface{}, callback func([]interface{})) {

	d.mu.Lock()

	if acc, exists := d.acc[key]; !exists {
		t := time.NewTimer(d.delay)
		d.acc[key] = &Accumulation{
			messages: []interface{}{message},
			t:        t,
		}
		go d.waitTimer(key, callback, t)
	} else {
		acc.messages = append(acc.messages, message)
		acc.t.Reset(d.delay)
	}

	d.mu.Unlock()
}

func (d *Accumulator) waitTimer(key string, callback func([]interface{}), t *time.Timer) {
	<-t.C
	d.mu.Lock()
	acc, exists := d.acc[key]
	if !exists {
		return
	}
	callback(acc.messages)
	d.acc[key] = nil
	d.mu.Unlock()
}