scheduler

Documentation

Index

• Variables
• type Command
  • func CommandWithExecuteTime(cmd eh.Command, t time.Time) Command
• type Error
  • func (e *Error) Cause() error
  • func (e *Error) Error() string
  • func (e *Error) Unwrap() error
• type PersistedCommand
  • func (c *PersistedCommand) EntityID() uuid.UUID
• type Scheduler
  • func NewMiddleware(repo eh.ReadWriteRepo, codec eh.CommandCodec) (eh.CommandHandlerMiddleware, *Scheduler)
  • func (s *Scheduler) CancelCommand(ctx context.Context, id uuid.UUID) error
  • func (s *Scheduler) Commands(ctx context.Context) ([]*PersistedCommand, error)
  • func (s *Scheduler) Errors() <-chan error
  • func (s *Scheduler) Load(ctx context.Context) error
  • func (s *Scheduler) ScheduleCommand(ctx context.Context, cmd eh.Command, executeAt time.Time) (uuid.UUID, error)
  • func (s *Scheduler) Start() error
  • func (s *Scheduler) Stop() error

Variables

var ErrCanceled = errors.New("canceled")

ErrCanceled is when a scheduled command has been canceled.

var ScheduledCommandsQueueSize = 100

The default command queue size to use.

Types

type Command

type Command interface {
	eh.Command

	// ExecuteAt returns the time when the command will execute.
	ExecuteAt() time.Time
}

Command is a scheduled command with an execution time.

func CommandWithExecuteTime

func CommandWithExecuteTime(cmd eh.Command, t time.Time) Command

CommandWithExecuteTime returns a wrapped command with a execution time set.

type Error

type Error struct {
	// Err is the error that happened when handling the command.
	Err error
	// Ctx is the context used when the error happened.
	Ctx context.Context
	// Command is the command handeled when the error happened.
	Command eh.Command
}

Error is an async error containing the error and the command.

func (*Error) Cause

func (e *Error) Cause() error

Cause implements the github.com/pkg/errors Unwrap method.

func (*Error) Error

func (e *Error) Error() string

Error implements the Error method of the error interface.

func (*Error) Unwrap

func (e *Error) Unwrap() error

Unwrap implements the errors.Unwrap method.

type PersistedCommand

type PersistedCommand struct {
	ID         uuid.UUID       `json:"_"                 bson:"_id"`
	IDStr      string          `json:"id"                bson:"_"`
	RawCommand []byte          `json:"command"           bson:"command"`
	ExecuteAt  time.Time       `json:"timestamp"         bson:"timestamp"`
	Command    eh.Command      `json:"-"                 bson:"-"`
	Context    context.Context `json:"-"                 bson:"-"`
}

PersistedCommand is a persisted command.

func (*PersistedCommand) EntityID

func (c *PersistedCommand) EntityID() uuid.UUID

EntityID implements the EntityID method of the eventhorizon.Entity interface.

type Scheduler

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

Scheduler is a scheduled of commands.

func NewMiddleware

func NewMiddleware(repo eh.ReadWriteRepo, codec eh.CommandCodec) (eh.CommandHandlerMiddleware, *Scheduler)

NewMiddleware returns a new command handler middleware and a scheduler helper.

func (*Scheduler) CancelCommand

func (s *Scheduler) CancelCommand(ctx context.Context, id uuid.UUID) error

CancelCommand cancels a scheduled command.

func (*Scheduler) Commands

func (s *Scheduler) Commands(ctx context.Context) ([]*PersistedCommand, error)

Commands returns all scheduled commands.

func (*Scheduler) Errors

func (s *Scheduler) Errors() <-chan error

Errors returns an error channel that will receive errors from handling of scheduled commands.

func (*Scheduler) Load

func (s *Scheduler) Load(ctx context.Context) error

Load loads all persisted scheduled commands. It will be limited by ScheduledCommandsQueueSize if Start() has not yet been called.

func (*Scheduler) ScheduleCommand

func (s *Scheduler) ScheduleCommand(ctx context.Context, cmd eh.Command, executeAt time.Time) (uuid.UUID, error)

ScheduleCommand schedules a command to be executed at `executeAt`. It is persisted to the repo.

func (*Scheduler) Start

func (s *Scheduler) Start() error

Start starts the scheduler.

func (*Scheduler) Stop

func (s *Scheduler) Stop() error

Stop stops all scheduled commands.