coalescing_queue.go

package coalescing

import (
	"context"
	"sync/atomic"
)

func Q(ctx context.Context, bufferSize int, fn func()) func() {
	var (
		q = make(chan int64, bufferSize)
		v int64
	)
	go func() {
		for {
			select {
			case n := <-q:
				if v > n {
					continue
				}
				fn()
			case <-ctx.Done():
				return
			}
		}
	}()

	return func() {
		q <- atomic.AddInt64(&v, 1)
	}
}