singleflight

Documentation

Index

• Constants
• Variables
• type BackOff
• type Cache
  • func NewCache[T any](client *redis.Client) *Cache[T]
  • func (c *Cache[T]) LoadOrStore(ctx context.Context, key string, getter func(ctx context.Context) (T, error), ...) (T, bool, error)
• type ExponentialBackOff
  • func NewExponentialBackOff(base, cap time.Duration) *ExponentialBackOff
  • func (b *ExponentialBackOff) Duration(i int) time.Duration
• type Group
  • func New(client *redis.Client) *Group
  • func (g *Group) Do(ctx context.Context, key string, fn func(context.Context) error, ...) (doOrWait bool, err error)

Examples

• Cache

Constants

const OK = "ok"

Variables

var (
	ErrTimeout            = errors.New("group: timeout waiting for result")
	ErrSubscriptionClosed = errors.New("group: subscription closed")
)

Types

type BackOff

type BackOff interface {
	Duration(i int) time.Duration
}

type Cache

type Cache[T any] struct {
	Client  *redis.Client
	Group   *Group
	LockTTL time.Duration
	WaitTTL time.Duration
	Suffix  string
}

Example

// Cache example.
// When the cache is stale, only once worker will populate
// the cache, and the rest will wait.

stop := redistest.Init()
defer stop()

client := redis.NewClient(&redis.Options{
	Addr: redistest.Addr(),
})
defer client.Close()

cache := singleflight.NewCache[string](client)
ctx := context.Background()

hit := new(atomic.Int64)
exec := new(atomic.Int64)

n := 10
var wg sync.WaitGroup
wg.Add(n)

for range n {
	go func() {
		defer wg.Done()
		v, hitOrMiss, err := cache.LoadOrStore(ctx, "foo", func(context.Context) (string, error) {
			exec.Add(1)
			return "bar", nil
		}, 10*time.Second)
		if err != nil {
			panic(err)
		}
		if v != "bar" {
			panic("unexpected value: " + v)
		}
		if hitOrMiss {
			hit.Add(1)
		}

	}()
}
wg.Wait()

fmt.Println("hit:", hit.Load())
fmt.Println("exec:", exec.Load())

Output:

hit: 9
exec: 1

func NewCache

func NewCache[T any](client *redis.Client) *Cache[T]

func (*Cache[T]) LoadOrStore

func (c *Cache[T]) LoadOrStore(ctx context.Context, key string, getter func(ctx context.Context) (T, error), ttl time.Duration) (T, bool, error)

type ExponentialBackOff

type ExponentialBackOff struct {
	Base time.Duration
	Cap  time.Duration
}

func NewExponentialBackOff

func NewExponentialBackOff(base, cap time.Duration) *ExponentialBackOff

func (*ExponentialBackOff) Duration

func (b *ExponentialBackOff) Duration(i int) time.Duration

type Group

type Group struct {
	BackOff BackOff
	Client  *redis.Client
	Locker  *lock.Locker
	Group   *singleflight.Group[bool]
}

func New

func New(client *redis.Client) *Group

func (*Group) Do

func (g *Group) Do(ctx context.Context, key string, fn func(context.Context) error, lockTTL, waitTTL time.Duration) (doOrWait bool, err error)