any-sync/commonspace/object/keyvalue/limiter.go

package keyvalue

import (
	"context"
	"sync"
)

type concurrentLimiter struct {
	mu         sync.Mutex
	inProgress map[string]bool
	wg         sync.WaitGroup
}

func newConcurrentLimiter() *concurrentLimiter {
	return &concurrentLimiter{
		inProgress: make(map[string]bool),
	}
}

func (cl *concurrentLimiter) ScheduleRequest(ctx context.Context, id string, action func()) bool {
	cl.mu.Lock()
	if cl.inProgress[id] {
		cl.mu.Unlock()
		return false
	}

	cl.inProgress[id] = true
	cl.wg.Add(1)
	cl.mu.Unlock()

	go func() {
		defer func() {
			cl.mu.Lock()
			delete(cl.inProgress, id)
			cl.mu.Unlock()
			cl.wg.Done()
		}()

		select {
		case <-ctx.Done():
			return
		default:
			action()
		}
	}()

	return true
}

func (cl *concurrentLimiter) Close() {
	cl.wg.Wait()
}