core

Documentation

Index

• Constants
• type Core
  • func New(o Opt, srcDBs dbpool.Pool, res ResultBackends, lo *slog.Logger) *Core
  • func (co *Core) CancelJob(jobID string, purge bool) error
  • func (co *Core) CancelJobGroup(groupID string, purge bool) error
  • func (co *Core) GetJobGroupStatus(groupID string) (models.GroupStatusResp, error)
  • func (co *Core) GetJobStatus(jobID string) (models.JobStatusResp, error)
  • func (co *Core) GetPendingJobs(queue string) ([]tasqueue.JobMessage, error)
  • func (co *Core) GetTasks() Tasks
  • func (co *Core) LoadTasks(dirs []string) error
  • func (co *Core) NewJob(j models.JobReq, taskName string) (models.JobResp, error)
  • func (co *Core) NewJobGroup(req models.GroupReq) (models.GroupResp, error)
  • func (co *Core) Start(ctx context.Context, workerName string, concurrency int) error
• type Opt
• type ResultBackends
  • func (r ResultBackends) Get(name string) models.ResultBackend
  • func (r ResultBackends) GetNames() []string
  • func (r ResultBackends) GetRandom() (string, models.ResultBackend)
• type Task
• type Tasks

Constants

const (
	StatusPending = "PENDING"
	StatusStarted = "STARTED"
	StatusFailure = "FAILURE"
	StatusSuccess = "SUCCESS"
	StatusRetry   = "RETRY"
)

Types

type Core

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

func New

func New(o Opt, srcDBs dbpool.Pool, res ResultBackends, lo *slog.Logger) *Core

New returns a new instance of Core.

func (*Core) CancelJob

func (co *Core) CancelJob(jobID string, purge bool) error

CancelJob cancels a given pending/processing job.

func (*Core) CancelJobGroup

func (co *Core) CancelJobGroup(groupID string, purge bool) error

CancelJobGroup cancels a pending group job by individually cancelling all pending jobs in it.

func (*Core) GetJobGroupStatus

func (co *Core) GetJobGroupStatus(groupID string) (models.GroupStatusResp, error)

GetJobGroupStatus returns the status of a job group including statuses of individual jobs in it.

func (*Core) GetJobStatus

func (co *Core) GetJobStatus(jobID string) (models.JobStatusResp, error)

GetJobStatus returns the status of a job.

func (*Core) GetPendingJobs

func (co *Core) GetPendingJobs(queue string) ([]tasqueue.JobMessage, error)

GetPendingJobs returns jobs pending execution.

func (*Core) GetTasks

func (co *Core) GetTasks() Tasks

GetTasks returns the registered tasks map.

func (*Core) LoadTasks

func (co *Core) LoadTasks(dirs []string) error

LoadTasks loads SQL queries from all the .sql files in a given directory.

func (*Core) NewJob

func (co *Core) NewJob(j models.JobReq, taskName string) (models.JobResp, error)

NewJob creates a new job out of a given task and registers it.

func (*Core) NewJobGroup

func (co *Core) NewJobGroup(req models.GroupReq) (models.GroupResp, error)

func (*Core) Start

func (co *Core) Start(ctx context.Context, workerName string, concurrency int) error

Start is a blocking function that spawns the queue workers and starts processing jobs. This should be invoked last after all other initializations are done.

type Opt

type Opt struct {
	DefaultQueue            string
	DefaultGroupConcurrency int
	DefaultJobTTL           time.Duration

	// DSNs for connecting to the broker backend and the broker state backend.
	Broker  tasqueue.Broker
	Results tasqueue.Results
}

Opt represents core options.

type ResultBackends

type ResultBackends map[string]models.ResultBackend

ResultBackends represents a map of result writing backends (sql DBs).

func (ResultBackends) Get

func (r ResultBackends) Get(name string) models.ResultBackend

Get returns an *sql.DB from the DBs map by name.

func (ResultBackends) GetNames

func (r ResultBackends) GetNames() []string

GetNames returns the list of available DB names.

func (ResultBackends) GetRandom

func (r ResultBackends) GetRandom() (string, models.ResultBackend)

GetRandom returns a random results backend from the map.

type Task

type Task struct {
	Name           string         `json:"name"`
	Queue          string         `json:"queue"`
	Conc           int            `json:"concurrency"`
	Stmt           *sql.Stmt      `json:"-"`
	Raw            string         `json:"raw,omitempty"`
	DBs            dbpool.Pool    `json:"-"`
	ResultBackends ResultBackends `json:"-"`
}

Task represents an SQL query with its prepared and raw forms.

type Tasks

type Tasks map[string]Task

Tasks represents a map of prepared SQL statements.