reconcile

Documentation

Index

• type BackoffResetCallbackFunc
• type BackoffStartCallbackFunc
• type Cache
  • func NewCache[K comparable, V any]() *Cache[K, V]
  • func (r *Cache[K, V]) Delete(key K)
  • func (r *Cache[K, V]) Freeze(key K, freeze bool)
  • func (r *Cache[K, V]) Get(key K) (old, latest optional.Value[V])
  • func (r *Cache[K, V]) Update(key K, old, latest *V)
• type CompareObjectHandleFunc
• type Core
  • func NewCore[K comparable, V any](ctx context.Context, resolvedOpts Options[K, V]) *Core[K, V]
  • func (c *Core[K, V]) CancelSchedule(job queue.Job[K]) bool
  • func (c *Core[K, V]) Reconcile(stop <-chan struct{})
  • func (c *Core[K, V]) Schedule(job queue.Job[K], delay time.Duration) error
  • func (c *Core[K, V]) Start() error
• type HandleFuncs
  • func (h *HandleFuncs[T]) ResolveNil() (result HandleFuncs[T])
• type Interface
• type Options
  • func (o Options[K, T]) ResolveNil() (ret Options[K, T])
• type Result
• type SingleObjectHandleFunc

Types

type BackoffResetCallbackFunc

type BackoffResetCallbackFunc[K comparable] func(key K)

type BackoffStartCallbackFunc

type BackoffStartCallbackFunc[K comparable] func(key K, reason error)

type Cache

type Cache[K comparable, V any] struct {
	// contains filtered or unexported fields
}

func NewCache

func NewCache[K comparable, V any]() *Cache[K, V]

func (*Cache[K, V]) Delete

func (r *Cache[K, V]) Delete(key K)

func (*Cache[K, V]) Freeze

func (r *Cache[K, V]) Freeze(key K, freeze bool)

Freeze object with key in old cache

func (*Cache[K, V]) Get

func (r *Cache[K, V]) Get(key K) (old, latest optional.Value[V])

func (*Cache[K, V]) Update

func (r *Cache[K, V]) Update(key K, old, latest *V)

type CompareObjectHandleFunc

type CompareObjectHandleFunc[T any] func(oldObj, newObj T) optional.Value[Result]

type Core

type Core[K comparable, V any] struct {
	*Cache[K, V]
	// contains filtered or unexported fields
}

func NewCore

func NewCore[K comparable, V any](
	ctx context.Context, resolvedOpts Options[K, V],
) *Core[K, V]

func (*Core[K, V]) CancelSchedule

func (c *Core[K, V]) CancelSchedule(job queue.Job[K]) bool

func (*Core[K, V]) Reconcile

func (c *Core[K, V]) Reconcile(stop <-chan struct{})

Reconcile jobs until stop released

func (*Core[K, V]) Schedule

func (c *Core[K, V]) Schedule(job queue.Job[K], delay time.Duration) error

func (*Core[K, V]) Start

func (c *Core[K, V]) Start() error

Start handling of delayed jobs

type HandleFuncs

type HandleFuncs[T any] struct {
	OnAdded    SingleObjectHandleFunc[T]
	OnUpdated  CompareObjectHandleFunc[T]
	OnDeleting SingleObjectHandleFunc[T]
	OnDeleted  SingleObjectHandleFunc[T]
}

func (*HandleFuncs[T]) ResolveNil

func (h *HandleFuncs[T]) ResolveNil() (result HandleFuncs[T])

type Interface

type Interface[T comparable] interface {
	Start() error
	ReconcileUntil(stop <-chan struct{})
	Schedule(job queue.Job[T], delay time.Duration) error
	CancelSchedule(job queue.Job[T]) bool
}

type Options

type Options[K comparable, T any] struct {
	Logger          log.Interface
	BackoffStrategy *backoff.Strategy[K]
	Workers         int
	RequireCache    bool
	Handlers        HandleFuncs[T]
	OnBackoffStart  BackoffStartCallbackFunc[K]
	OnBackoffReset  BackoffResetCallbackFunc[K]
}

func (Options[K, T]) ResolveNil

func (o Options[K, T]) ResolveNil() (ret Options[K, T])

type Result

type Result struct {
	NextAction    queue.JobAction
	ScheduleAfter time.Duration
	Err           error
}

type SingleObjectHandleFunc

type SingleObjectHandleFunc[T any] func(obj T) optional.Value[Result]