Default to SQLite rather than Postgres & Redis (#72)
* Initial rough draft switch to sqlite. * Rewrote cache implemenation. * Provide typed tasks. * Task cleanup. * Use same db for tasks. * Provide task queue registration and service container injection. * Added optional delay to tasks. Pool buffers when encoding. * Added tests for the task client and runner. * Added handler examples for caching and tasks. * Cleanup and documentation. * Use make in workflow. * Updated documentation. * Updated documentation.
This commit is contained in:
Mike Stefanello
GitHub
parent
5e9e502b42
commit
a096abd195
29 changed files with 956 additions and 910 deletions
|
|
@ -1,179 +1,204 @@
|
|||
package services
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/gob"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/hibiken/asynq"
|
||||
"github.com/maragudk/goqite"
|
||||
"github.com/maragudk/goqite/jobs"
|
||||
"github.com/mikestefanello/pagoda/config"
|
||||
"github.com/mikestefanello/pagoda/pkg/log"
|
||||
)
|
||||
|
||||
type (
|
||||
// TaskClient is that client that allows you to queue or schedule task execution
|
||||
// TaskClient is that client that allows you to queue or schedule task execution.
|
||||
// Under the hood we create only a single queue using goqite for all tasks because we do not want more than one
|
||||
// runner to process the tasks. The TaskClient wrapper provides abstractions for separate, type-safe queues.
|
||||
TaskClient struct {
|
||||
// client stores the asynq client
|
||||
client *asynq.Client
|
||||
|
||||
// scheduler stores the asynq scheduler
|
||||
scheduler *asynq.Scheduler
|
||||
queue *goqite.Queue
|
||||
runner *jobs.Runner
|
||||
buffers sync.Pool
|
||||
}
|
||||
|
||||
// task handles task creation operations
|
||||
task struct {
|
||||
client *TaskClient
|
||||
typ string
|
||||
payload any
|
||||
periodic *string
|
||||
queue *string
|
||||
maxRetries *int
|
||||
timeout *time.Duration
|
||||
deadline *time.Time
|
||||
at *time.Time
|
||||
wait *time.Duration
|
||||
retain *time.Duration
|
||||
// Task is a job that can be added to a queue and later passed to and executed by a QueueSubscriber.
|
||||
// See pkg/tasks for an example of how this can be used with a queue.
|
||||
Task interface {
|
||||
Name() string
|
||||
}
|
||||
|
||||
// TaskSaveOp handles task save operations
|
||||
TaskSaveOp struct {
|
||||
client *TaskClient
|
||||
task Task
|
||||
tx *sql.Tx
|
||||
at *time.Time
|
||||
wait *time.Duration
|
||||
}
|
||||
|
||||
// Queue is a queue that a Task can be pushed to for execution.
|
||||
// While this can be implemented directly, it's recommended to use NewQueue() which uses generics in
|
||||
// order to provide type-safe queues and queue subscriber callbacks for task execution.
|
||||
Queue interface {
|
||||
// Name returns the name of the task this queue processes
|
||||
Name() string
|
||||
|
||||
// Receive receives the Task payload to be processed
|
||||
Receive(ctx context.Context, payload []byte) error
|
||||
}
|
||||
|
||||
// queue provides a type-safe implementation of Queue
|
||||
queue[T Task] struct {
|
||||
name string
|
||||
subscriber QueueSubscriber[T]
|
||||
}
|
||||
|
||||
// QueueSubscriber is a generic subscriber callback for a given queue to process Tasks
|
||||
QueueSubscriber[T Task] func(context.Context, T) error
|
||||
)
|
||||
|
||||
// NewTaskClient creates a new task client
|
||||
func NewTaskClient(cfg *config.Config) *TaskClient {
|
||||
// Determine the database based on the environment
|
||||
db := cfg.Cache.Database
|
||||
if cfg.App.Environment == config.EnvTest {
|
||||
db = cfg.Cache.TestDatabase
|
||||
func NewTaskClient(cfg config.TasksConfig, db *sql.DB) (*TaskClient, error) {
|
||||
// Install the schema
|
||||
if err := goqite.Setup(context.Background(), db); err != nil {
|
||||
// An error is returned if we already ran this and there's no better way to check.
|
||||
// You can and probably should handle this via migrations
|
||||
if !strings.Contains(err.Error(), "already exists") {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
conn := asynq.RedisClientOpt{
|
||||
Addr: fmt.Sprintf("%s:%d", cfg.Cache.Hostname, cfg.Cache.Port),
|
||||
Password: cfg.Cache.Password,
|
||||
DB: db,
|
||||
t := &TaskClient{
|
||||
queue: goqite.New(goqite.NewOpts{
|
||||
DB: db,
|
||||
Name: "tasks",
|
||||
MaxReceive: cfg.MaxRetries,
|
||||
}),
|
||||
buffers: sync.Pool{
|
||||
New: func() interface{} {
|
||||
return bytes.NewBuffer(nil)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
return &TaskClient{
|
||||
client: asynq.NewClient(conn),
|
||||
scheduler: asynq.NewScheduler(conn, nil),
|
||||
}
|
||||
t.runner = jobs.NewRunner(jobs.NewRunnerOpts{
|
||||
Limit: cfg.Goroutines,
|
||||
Log: log.Default(),
|
||||
PollInterval: cfg.PollInterval,
|
||||
Queue: t.queue,
|
||||
})
|
||||
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// Close closes the connection to the task service
|
||||
func (t *TaskClient) Close() error {
|
||||
return t.client.Close()
|
||||
// StartRunner starts the scheduler service which adds scheduled tasks to the queue.
|
||||
// This must be running in order to execute queued tasked.
|
||||
// To stop the runner, cancel the context.
|
||||
// This is a blocking call.
|
||||
func (t *TaskClient) StartRunner(ctx context.Context) {
|
||||
t.runner.Start(ctx)
|
||||
}
|
||||
|
||||
// StartScheduler starts the scheduler service which adds scheduled tasks to the queue
|
||||
// This must be running in order to queue tasks set for periodic execution
|
||||
func (t *TaskClient) StartScheduler() error {
|
||||
return t.scheduler.Run()
|
||||
// Register registers a queue so tasks can be added to it and processed
|
||||
func (t *TaskClient) Register(queue Queue) {
|
||||
t.runner.Register(queue.Name(), queue.Receive)
|
||||
}
|
||||
|
||||
// New starts a task creation operation
|
||||
func (t *TaskClient) New(typ string) *task {
|
||||
return &task{
|
||||
// New starts a task save operation
|
||||
func (t *TaskClient) New(task Task) *TaskSaveOp {
|
||||
return &TaskSaveOp{
|
||||
client: t,
|
||||
typ: typ,
|
||||
task: task,
|
||||
}
|
||||
}
|
||||
|
||||
// Payload sets the task payload data which will be sent to the task handler
|
||||
func (t *task) Payload(payload any) *task {
|
||||
t.payload = payload
|
||||
return t
|
||||
}
|
||||
|
||||
// Periodic sets the task to execute periodically according to a given interval
|
||||
// The interval can be either in cron form ("*/5 * * * *") or "@every 30s"
|
||||
func (t *task) Periodic(interval string) *task {
|
||||
t.periodic = &interval
|
||||
return t
|
||||
}
|
||||
|
||||
// Queue specifies the name of the queue to add the task to
|
||||
// The default queue will be used if this is not set
|
||||
func (t *task) Queue(queue string) *task {
|
||||
t.queue = &queue
|
||||
return t
|
||||
}
|
||||
|
||||
// Timeout sets the task timeout, meaning the task must execute within a given duration
|
||||
func (t *task) Timeout(timeout time.Duration) *task {
|
||||
t.timeout = &timeout
|
||||
return t
|
||||
}
|
||||
|
||||
// Deadline sets the task execution deadline to a specific date and time
|
||||
func (t *task) Deadline(deadline time.Time) *task {
|
||||
t.deadline = &deadline
|
||||
return t
|
||||
}
|
||||
|
||||
// At sets the exact date and time the task should be executed
|
||||
func (t *task) At(processAt time.Time) *task {
|
||||
t.at = &processAt
|
||||
func (t *TaskSaveOp) At(processAt time.Time) *TaskSaveOp {
|
||||
t.Wait(time.Until(processAt))
|
||||
return t
|
||||
}
|
||||
|
||||
// Wait instructs the task to wait a given duration before it is executed
|
||||
func (t *task) Wait(duration time.Duration) *task {
|
||||
func (t *TaskSaveOp) Wait(duration time.Duration) *TaskSaveOp {
|
||||
t.wait = &duration
|
||||
return t
|
||||
}
|
||||
|
||||
// Retain instructs the task service to retain the task data for a given duration after execution is complete
|
||||
func (t *task) Retain(duration time.Duration) *task {
|
||||
t.retain = &duration
|
||||
// Tx will include the task as part of a given database transaction
|
||||
func (t *TaskSaveOp) Tx(tx *sql.Tx) *TaskSaveOp {
|
||||
t.tx = tx
|
||||
return t
|
||||
}
|
||||
|
||||
// MaxRetries sets the maximum amount of times to retry executing the task in the event of a failure
|
||||
func (t *task) MaxRetries(retries int) *task {
|
||||
t.maxRetries = &retries
|
||||
return t
|
||||
}
|
||||
|
||||
// Save saves the task so it can be executed
|
||||
func (t *task) Save() error {
|
||||
var err error
|
||||
|
||||
// Build the payload
|
||||
var payload []byte
|
||||
if t.payload != nil {
|
||||
if payload, err = json.Marshal(t.payload); err != nil {
|
||||
return err
|
||||
}
|
||||
// Save saves the task, so it can be queued for execution
|
||||
func (t *TaskSaveOp) Save() error {
|
||||
type message struct {
|
||||
Name string
|
||||
Message []byte
|
||||
}
|
||||
|
||||
// Build the task options
|
||||
opts := make([]asynq.Option, 0)
|
||||
if t.queue != nil {
|
||||
opts = append(opts, asynq.Queue(*t.queue))
|
||||
// Encode the task
|
||||
taskBuf := t.client.buffers.Get().(*bytes.Buffer)
|
||||
if err := gob.NewEncoder(taskBuf).Encode(t.task); err != nil {
|
||||
return err
|
||||
}
|
||||
if t.maxRetries != nil {
|
||||
opts = append(opts, asynq.MaxRetry(*t.maxRetries))
|
||||
|
||||
// Wrap and encode the message
|
||||
// This is needed as a workaround because goqite doesn't support delays using the jobs package,
|
||||
// so we format the message the way it expects but use the queue to supply the delay
|
||||
msgBuf := t.client.buffers.Get().(*bytes.Buffer)
|
||||
wrapper := message{Name: t.task.Name(), Message: taskBuf.Bytes()}
|
||||
if err := gob.NewEncoder(msgBuf).Encode(wrapper); err != nil {
|
||||
return err
|
||||
}
|
||||
if t.timeout != nil {
|
||||
opts = append(opts, asynq.Timeout(*t.timeout))
|
||||
}
|
||||
if t.deadline != nil {
|
||||
opts = append(opts, asynq.Deadline(*t.deadline))
|
||||
|
||||
msg := goqite.Message{
|
||||
Body: msgBuf.Bytes(),
|
||||
}
|
||||
|
||||
if t.wait != nil {
|
||||
opts = append(opts, asynq.ProcessIn(*t.wait))
|
||||
}
|
||||
if t.retain != nil {
|
||||
opts = append(opts, asynq.Retention(*t.retain))
|
||||
}
|
||||
if t.at != nil {
|
||||
opts = append(opts, asynq.ProcessAt(*t.at))
|
||||
msg.Delay = *t.wait
|
||||
}
|
||||
|
||||
// Build the task
|
||||
task := asynq.NewTask(t.typ, payload, opts...)
|
||||
// Put the buffers back in the pool for re-use
|
||||
taskBuf.Reset()
|
||||
msgBuf.Reset()
|
||||
t.client.buffers.Put(taskBuf)
|
||||
t.client.buffers.Put(msgBuf)
|
||||
|
||||
// Schedule, if needed
|
||||
if t.periodic != nil {
|
||||
_, err = t.client.scheduler.Register(*t.periodic, task)
|
||||
if t.tx == nil {
|
||||
return t.client.queue.Send(context.Background(), msg)
|
||||
} else {
|
||||
_, err = t.client.client.Enqueue(task)
|
||||
return t.client.queue.SendTx(context.Background(), t.tx, msg)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// NewQueue queues a new type-safe Queue of a given Task type
|
||||
func NewQueue[T Task](subscriber QueueSubscriber[T]) Queue {
|
||||
var task T
|
||||
|
||||
q := &queue[T]{
|
||||
name: task.Name(),
|
||||
subscriber: subscriber,
|
||||
}
|
||||
|
||||
return q
|
||||
}
|
||||
|
||||
func (q *queue[T]) Name() string {
|
||||
return q.name
|
||||
}
|
||||
|
||||
func (q *queue[T]) Receive(ctx context.Context, payload []byte) error {
|
||||
var obj T
|
||||
err := gob.NewDecoder(bytes.NewReader(payload)).Decode(&obj)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return q.subscriber(ctx, obj)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue