projection

Documentation

Index

• Variables
• func Deduplicate[T comparable](in <-chan T, window time.Duration) <-chan T
• func Filter[T ProgressAwareTarget](t T, evts []event.Event) ([]event.Event, error)
• func PeriodicJobTicker(ctx context.Context, d time.Duration, s projection.Schedule, ...) <-chan error
• func PeriodicJobTickerFunc(ctx context.Context, d time.Duration, s projection.Schedule, ...) <-chan error
• func Use[T ProgressAwareTarget](j projection.Job, newFn func(context.Context) (T, error), ...) error
• type ContextKey
• type Group
  • func GroupByFunc[K comparable, V any](ctx context.Context, groupFn func(V) (K, error), in <-chan V, ...) ([]Group[K, V], error)
• type Groups
  • func (g Groups[K, V]) Keys() []K
  • func (g Groups[K, V]) Map() map[K]Group[K, V]
• type ProgressAwareTarget
• type ProjectionProgressor
  • func NewProgressor() *ProjectionProgressor
  • func (p *ProjectionProgressor) Dirty() bool
  • func (p *ProjectionProgressor) SetProgress(t time.Time, ids ...uuid.UUID)
• type Projector
• type Registerer
• type Scheduler
  • func (s *Scheduler) Subscribe(ctx context.Context, handler func(projection.Job) error) (<-chan error, error)
• type Schedules
• type Starter
• type Worker
  • func New(projector Projector, bus event.Bus, reg *codec.Registry, opts ...WorkerOption) *Worker
  • func (svc *Worker) Start(ctx context.Context, opts ...projection.SubscribeOption) (<-chan error, error)
• type WorkerOption
  • func WithErrorFunc(fn func(projection.Job, error)) WorkerOption
  • func WithLogger(logger *zap.SugaredLogger) WorkerOption

Variables

var (
	ContextKeyJobName = ContextKey("name")
	ContextKeyId      = ContextKey("id")
	ContextKeyRunId   = ContextKey("run-id")
)

var (
	ErrNotFound = errors.New("projection not found")
)

Functions

func Deduplicate

func Deduplicate[T comparable](in <-chan T, window time.Duration) <-chan T

func Filter

func Filter[T ProgressAwareTarget](t T, evts []event.Event) ([]event.Event, error)

func PeriodicJobTicker

func PeriodicJobTicker(ctx context.Context, d time.Duration, s projection.Schedule, opts ...query.Option) <-chan error

func PeriodicJobTickerFunc

func PeriodicJobTickerFunc(ctx context.Context, d time.Duration, s projection.Schedule, tickFn func(context.Context) ([]query.Option, error), opts ...query.Option) <-chan error

func Use

func Use[T ProgressAwareTarget](
	j projection.Job,
	newFn func(context.Context) (T, error),
	getFn func(context.Context, string) (T, error),
	saveFn func(context.Context, T) error,
	useFn func(context.Context, T) error) error

Types

type ContextKey

type ContextKey string

type Group

type Group[K comparable, V any] struct {
	Key   K
	Value []V
}

func GroupByFunc

func GroupByFunc[K comparable, V any](ctx context.Context, groupFn func(V) (K, error), in <-chan V, errs ...<-chan error) ([]Group[K, V], error)

type Groups

type Groups[K comparable, V any] []Group[K, V]

func (Groups[K, V]) Keys

func (g Groups[K, V]) Keys() []K

func (Groups[K, V]) Map

func (g Groups[K, V]) Map() map[K]Group[K, V]

type ProgressAwareTarget

type ProgressAwareTarget interface {
	ApplyEvent(event.Event)
	Progress() (time.Time, []uuid.UUID)
	Key() string
	Dirty() bool
}

type ProjectionProgressor

type ProjectionProgressor struct {
	*projection.Progressor
	// contains filtered or unexported fields
}

func NewProgressor

func NewProgressor() *ProjectionProgressor

func (*ProjectionProgressor) Dirty

func (p *ProjectionProgressor) Dirty() bool

func (*ProjectionProgressor) SetProgress

func (p *ProjectionProgressor) SetProgress(t time.Time, ids ...uuid.UUID)

type Projector

type Projector interface {
	Schedules() Schedules
	HandleJob(projection.Job) error
}

type Registerer

type Registerer interface {
	// RegisterEventHandler registers an event handler for the given event name.
	RegisterEventHandler(eventName string, handler func(event.Event) error)
}

type Scheduler

type Scheduler struct {
	Schedule       projection.Schedule
	StartupOptions []projection.SubscribeOption
}

func (*Scheduler) Subscribe

func (s *Scheduler) Subscribe(ctx context.Context, handler func(projection.Job) error) (<-chan error, error)

type Schedules

type Schedules = map[string]*Scheduler

type Starter

type Starter interface {
	Start(context.Context) error
}

type Worker

type Worker struct {
	// contains filtered or unexported fields
}

func New

func New(projector Projector, bus event.Bus, reg *codec.Registry, opts ...WorkerOption) *Worker

func (*Worker) Start

func (svc *Worker) Start(ctx context.Context, opts ...projection.SubscribeOption) (<-chan error, error)

type WorkerOption

type WorkerOption func(*Worker)

func WithErrorFunc

func WithErrorFunc(fn func(projection.Job, error)) WorkerOption

func WithLogger

func WithLogger(logger *zap.SugaredLogger) WorkerOption