task

Documentation

Index

• Constants
• type PostgresTaskRepository
  • func (r *PostgresTaskRepository) Create(tx *sql.Tx, t *Task) error
  • func (r *PostgresTaskRepository) CreateTable(tx *sql.Tx) error
  • func (r *PostgresTaskRepository) DeleteAll(tx *sql.Tx) error
  • func (r *PostgresTaskRepository) DeleteOld(tx *sql.Tx, moment time.Time) (int, error)
  • func (r *PostgresTaskRepository) FindNext(tx *sql.Tx) (*Task, error)
  • func (r *PostgresTaskRepository) Save(tx *sql.Tx, t *Task) error
• type Status
• type Task
  • func (e Task) LogValue() slog.Value
• type TaskFunc
• type TaskHandler
• type TaskQueryConfig
• type TaskRegistry
  • func NewTaskRegistry() *TaskRegistry
  • func (r *TaskRegistry) ApplyAsync(tx *sql.Tx, name string, args any, countdown *int, eta *time.Time) error
  • func (r *TaskRegistry) Call(tx *sql.Tx, name string, args any) (any, error)
  • func (r *TaskRegistry) Delay(tx *sql.Tx, name string, args any, countdown *int, eta *time.Time) error
  • func (r *TaskRegistry) Register(name string, f TaskFunc)
  • func (r *TaskRegistry) Unregister(name string)
• type TaskRepository
  • func NewPostgresTaskRepository(config ...TaskQueryConfig) TaskRepository
• type Worker
  • func NewWorker(db *sql.DB, config *WorkerConfig) *Worker
  • func (w *Worker) Start()
  • func (w *Worker) StartContext(ctx context.Context)
  • func (w *Worker) Stop()
• type WorkerConfig

Constants

const DefaultArgsField = "args"

const DefaultErrorField = "error"

const DefaultFinishedAtField = "finished_at"

const DefaultFuncField = "func"

const DefaultIdField = "id"

const DefaultResultField = "result"

const DefaultRunAtField = "run_at"

const DefaultStartedAtField = "started_at"

const DefaultStatusField = "status"

const DefaultTableName = "barn_task"

Types

type PostgresTaskRepository

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

func (*PostgresTaskRepository) Create

func (r *PostgresTaskRepository) Create(tx *sql.Tx, t *Task) error

func (*PostgresTaskRepository) CreateTable

func (r *PostgresTaskRepository) CreateTable(tx *sql.Tx) error

func (*PostgresTaskRepository) DeleteAll

func (r *PostgresTaskRepository) DeleteAll(tx *sql.Tx) error

func (*PostgresTaskRepository) DeleteOld

func (r *PostgresTaskRepository) DeleteOld(tx *sql.Tx, moment time.Time) (int, error)

func (*PostgresTaskRepository) FindNext

func (r *PostgresTaskRepository) FindNext(tx *sql.Tx) (*Task, error)

func (*PostgresTaskRepository) Save

func (r *PostgresTaskRepository) Save(tx *sql.Tx, t *Task) error

type Status

type Status string

const (
	Queued Status = "Q"
	Done   Status = "D"
	Failed Status = "F"
)

type Task

type Task struct {
	Id         int
	RunAt      time.Time
	Func       string
	Args       any
	Status     Status
	StartedAt  *time.Time
	FinishedAt *time.Time
	Result     any
	Error      *string
}

func (Task) LogValue

func (e Task) LogValue() slog.Value

type TaskFunc

type TaskFunc func(tx *sql.Tx, args any) (any, error)

type TaskHandler

type TaskHandler func(tx *sql.Tx, task *Task) error

type TaskQueryConfig

type TaskQueryConfig struct {
	TableName       string
	IdField         string
	RunAtField      string
	FuncField       string
	ArgsField       string
	StatusField     string
	StartedAtField  string
	FinishedAtField string
	ResultField     string
	ErrorField      string
}

type TaskRegistry

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

func NewTaskRegistry

func NewTaskRegistry() *TaskRegistry

func (*TaskRegistry) ApplyAsync

func (r *TaskRegistry) ApplyAsync(tx *sql.Tx, name string, args any, countdown *int, eta *time.Time) error

func (*TaskRegistry) Call

func (r *TaskRegistry) Call(tx *sql.Tx, name string, args any) (any, error)

func (*TaskRegistry) Delay

func (r *TaskRegistry) Delay(tx *sql.Tx, name string, args any, countdown *int, eta *time.Time) error

func (*TaskRegistry) Register

func (r *TaskRegistry) Register(name string, f TaskFunc)

func (*TaskRegistry) Unregister

func (r *TaskRegistry) Unregister(name string)

type TaskRepository

type TaskRepository interface {
	FindNext(tx *sql.Tx) (*Task, error)
	Create(tx *sql.Tx, task *Task) error
	Save(tx *sql.Tx, task *Task) error
	DeleteOld(tx *sql.Tx, t time.Time) (int, error)
}

func NewPostgresTaskRepository

func NewPostgresTaskRepository(config ...TaskQueryConfig) TaskRepository

type Worker

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

func NewWorker

func NewWorker(db *sql.DB, config *WorkerConfig) *Worker

func (*Worker) Start

func (w *Worker) Start()

func (*Worker) StartContext

func (w *Worker) StartContext(ctx context.Context)

func (*Worker) Stop

func (w *Worker) Stop()

type WorkerConfig

type WorkerConfig struct {
	Log        *slog.Logger
	Repository TaskRepository
	Cron       string
	Handler    TaskHandler
}