store

Documentation

Index

• Variables
• type OnDeleteFn
• type OnNotFoundFn
• type OnUpdateFn
• type Option
  • func WithJetstreamKV[T any, U protoutils.ProtoMessageWithMerge[T]](kv jetstream.KeyValue) Option[T, U]
  • func WithKVPrefix[T any, U protoutils.ProtoMessageWithMerge[T]](prefix string) Option[T, U]
  • func WithLocks[T any, U protoutils.ProtoMessageWithMerge[T]](l *locks.Locks) Option[T, U]
  • func WithOnDeleteFn[T any, U protoutils.ProtoMessageWithMerge[T]](fn OnDeleteFn[T, U]) Option[T, U]
  • func WithOnNotFoundFn[T any, U protoutils.ProtoMessageWithMerge[T]](fn OnNotFoundFn[T, U]) Option[T, U]
  • func WithOnUpdateFn[T any, U protoutils.ProtoMessageWithMerge[T]](fn OnUpdateFn[T, U]) Option[T, U]
• type Store
  • func New[T any, U protoutils.ProtoMessageWithMerge[T]](ctx context.Context, logger *zap.Logger, js *events.JSWrapper, bucket string, ...) (*Store[T, U], error)
  • func (s *Store[T, U]) Clear(ctx context.Context) error
  • func (s *Store[T, U]) ComputeUpdate(ctx context.Context, key string, load bool, ...) error
  • func (s *Store[T, U]) Delete(ctx context.Context, key string) error
  • func (s *Store[T, U]) Get(key string) (U, bool)
  • func (s *Store[T, U]) GetOrLoad(ctx context.Context, key string) (U, error)
  • func (s *Store[T, U]) Keys(ctx context.Context, prefix string) ([]string, error)
  • func (s *Store[T, U]) List() ([]U, error)
  • func (s *Store[T, U]) Load(ctx context.Context, key string) (U, error)
  • func (s *Store[T, U]) Put(ctx context.Context, key string, msg U) error
  • func (s *Store[T, U]) Range(ctx context.Context, fn func(key string, value U) bool) error
  • func (s *Store[T, U]) Start(ctx context.Context) error

Variables

var ErrNotFound = errors.New("store: not found")

Types

type OnDeleteFn

type OnDeleteFn[T any, U protoutils.ProtoMessageWithMerge[T]] func(s *Store[T, U], entry jetstream.KeyValueEntry, value U) error

type OnNotFoundFn

type OnNotFoundFn[T any, U protoutils.ProtoMessageWithMerge[T]] func(s *Store[T, U], ctx context.Context, key string) (U, error)

type OnUpdateFn

type OnUpdateFn[T any, U protoutils.ProtoMessageWithMerge[T]] func(s *Store[T, U], value U) (U, error)

type Option

type Option[T any, U protoutils.ProtoMessageWithMerge[T]] func(s *Store[T, U])

func WithJetstreamKV

func WithJetstreamKV[T any, U protoutils.ProtoMessageWithMerge[T]](kv jetstream.KeyValue) Option[T, U]

func WithKVPrefix

func WithKVPrefix[T any, U protoutils.ProtoMessageWithMerge[T]](prefix string) Option[T, U]

func WithLocks

func WithLocks[T any, U protoutils.ProtoMessageWithMerge[T]](l *locks.Locks) Option[T, U]

func WithOnDeleteFn

func WithOnDeleteFn[T any, U protoutils.ProtoMessageWithMerge[T]](fn OnDeleteFn[T, U]) Option[T, U]

func WithOnNotFoundFn

func WithOnNotFoundFn[T any, U protoutils.ProtoMessageWithMerge[T]](fn OnNotFoundFn[T, U]) Option[T, U]

func WithOnUpdateFn

func WithOnUpdateFn[T any, U protoutils.ProtoMessageWithMerge[T]](fn OnUpdateFn[T, U]) Option[T, U]

type Store

type Store[T any, U protoutils.ProtoMessageWithMerge[T]] struct {
	// contains filtered or unexported fields
}

func New

func New[T any, U protoutils.ProtoMessageWithMerge[T]](ctx context.Context, logger *zap.Logger, js *events.JSWrapper, bucket string, opts ...Option[T, U]) (*Store[T, U], error)

func (*Store[T, U]) Clear

func (s *Store[T, U]) Clear(ctx context.Context) error

func (*Store[T, U]) ComputeUpdate

func (s *Store[T, U]) ComputeUpdate(ctx context.Context, key string, load bool, fn func(key string, existing U) (U, bool, error)) error

func (*Store[T, U]) Delete

func (s *Store[T, U]) Delete(ctx context.Context, key string) error

func (*Store[T, U]) Get

func (s *Store[T, U]) Get(key string) (U, bool)

Get copy of data from local data

func (*Store[T, U]) GetOrLoad

func (s *Store[T, U]) GetOrLoad(ctx context.Context, key string) (U, error)

func (*Store[T, U]) Keys

func (s *Store[T, U]) Keys(ctx context.Context, prefix string) ([]string, error)

func (*Store[T, U]) List

func (s *Store[T, U]) List() ([]U, error)

func (*Store[T, U]) Load

func (s *Store[T, U]) Load(ctx context.Context, key string) (U, error)

Load data from kv store (this will add/update any existing local data entry) If no key is found, the original nats error is returned.

func (*Store[T, U]) Put

func (s *Store[T, U]) Put(ctx context.Context, key string, msg U) error

Put upload the message to kv and local

func (*Store[T, U]) Range

func (s *Store[T, U]) Range(ctx context.Context, fn func(key string, value U) bool) error

func (*Store[T, U]) Start

func (s *Store[T, U]) Start(ctx context.Context) error