Added asynq and a task client to the container to faciliate task queues.

services/container.go

@@ -46,6 +46,9 @@ type Container struct {
 
 	// TemplateRenderer stores a service to easily render and cache templates
 	TemplateRenderer *TemplateRenderer
+
+	// Tasks stores the task client
+	Tasks *TaskClient
 }
 
 // NewContainer creates and initializes a new Container
@@ -60,6 +63,7 @@ func NewContainer() *Container {
 	c.initAuth()
 	c.initTemplateRenderer()
 	c.initMail()
+	c.initTasks()
 	return c
 }
 
@@ -74,6 +78,9 @@ func (c *Container) Shutdown() error {
 	if err := c.Database.Close(); err != nil {
 		return err
 	}
+	if err := c.Tasks.Close(); err != nil {
+		return err
+	}
 
 	return nil
 }
@@ -182,3 +189,8 @@ func (c *Container) initMail() {
 		panic(fmt.Sprintf("failed to create mail client: %v", err))
 	}
 }
+
+// initTasks initializes the task client
+func (c *Container) initTasks() {
+	c.Tasks = NewTaskClient(c.Config.Cache)
+}

services/container_test.go

@@ -16,4 +16,5 @@ func TestNewContainer(t *testing.T) {
 	assert.NotNil(t, c.Mail)
 	assert.NotNil(t, c.Auth)
 	assert.NotNil(t, c.TemplateRenderer)
+	assert.NotNil(t, c.Tasks)
 }

services/services_test.go

@@ -23,11 +23,6 @@ func TestMain(m *testing.M) {
 
 	// Create a new container
 	c = NewContainer()
-	defer func() {
-		if err := c.Shutdown(); err != nil {
-			c.Web.Logger.Fatal(err)
-		}
-	}()
 
 	// Create a web context
 	ctx, _ = tests.NewContext(c.Web, "/")
@@ -41,5 +36,11 @@ func TestMain(m *testing.M) {
 
 	// Run tests
 	exitVal := m.Run()
+
+	// Shutdown the container
+	if err = c.Shutdown(); err != nil {
+		panic(err)
+	}
+
 	os.Exit(exitVal)
 }

services/tasks.go

@@ -0,0 +1,172 @@
+package services
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/mikestefanello/pagoda/config"
+)
+
+type (
+	// TaskClient is that client that allows you to queue or schedule task execution
+	TaskClient struct {
+		// client stores the asynq client
+		client *asynq.Client
+
+		// scheduler stores the asynq scheduler
+		scheduler *asynq.Scheduler
+	}
+
+	// task handles task creation operations
+	task struct {
+		client     *TaskClient
+		name       string
+		payload    interface{}
+		periodic   *string
+		queue      *string
+		maxRetries *int
+		timeout    *time.Duration
+		deadline   *time.Time
+		at         *time.Time
+		wait       *time.Duration
+		retain     *time.Duration
+	}
+)
+
+// NewTaskClient creates a new task client
+func NewTaskClient(cfg config.CacheConfig) *TaskClient {
+	conn := asynq.RedisClientOpt{
+		Addr:     fmt.Sprintf("%s:%d", cfg.Hostname, cfg.Port),
+		Password: cfg.Password,
+	}
+
+	return &TaskClient{
+		client:    asynq.NewClient(conn),
+		scheduler: asynq.NewScheduler(conn, nil),
+	}
+}
+
+// Close closes the connection to the task service
+func (t *TaskClient) Close() error {
+	return t.client.Close()
+}
+
+// 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()
+}
+
+// New starts a task creation operation
+func (t *TaskClient) New(name string) *task {
+	return &task{
+		client: t,
+		name:   name,
+	}
+}
+
+// Payload sets the task payload data which will be sent to the task handler
+func (t *task) Payload(payload interface{}) *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
+	return t
+}
+
+// Wait instructs the task to wait a given duration before it is executed
+func (t *task) Wait(duration time.Duration) *task {
+	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
+	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
+		}
+	}
+
+	// Build the task options
+	opts := make([]asynq.Option, 0)
+	if t.queue != nil {
+		opts = append(opts, asynq.Queue(*t.queue))
+	}
+	if t.maxRetries != nil {
+		opts = append(opts, asynq.MaxRetry(*t.maxRetries))
+	}
+	if t.timeout != nil {
+		opts = append(opts, asynq.Timeout(*t.timeout))
+	}
+	if t.deadline != nil {
+		opts = append(opts, asynq.Deadline(*t.deadline))
+	}
+	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))
+	}
+
+	// Build the task
+	task := asynq.NewTask(t.name, payload, opts...)
+
+	// Schedule, if needed
+	if t.periodic != nil {
+		_, err = t.client.scheduler.Register(*t.periodic, task)
+	} else {
+		_, err = t.client.client.Enqueue(task)
+	}
+	return err
+}

services/tasks_test.go

@@ -0,0 +1,35 @@
+package services
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTaskClient_New(t *testing.T) {
+	now := time.Now()
+	tk := c.Tasks.
+		New("task1").
+		Payload("payload").
+		Queue("queue").
+		Periodic("@every 5s").
+		MaxRetries(5).
+		Timeout(5 * time.Second).
+		Deadline(now).
+		At(now).
+		Wait(6 * time.Second).
+		Retain(7 * time.Second)
+
+	assert.Equal(t, "task1", tk.name)
+	assert.Equal(t, "payload", tk.payload.(string))
+	assert.Equal(t, "queue", *tk.queue)
+	assert.Equal(t, "@every 5s", *tk.periodic)
+	assert.Equal(t, 5, *tk.maxRetries)
+	assert.Equal(t, 5*time.Second, *tk.timeout)
+	assert.Equal(t, now, *tk.deadline)
+	assert.Equal(t, now, *tk.at)
+	assert.Equal(t, 6*time.Second, *tk.wait)
+	assert.Equal(t, 7*time.Second, *tk.retain)
+	assert.NoError(t, tk.Save())
+}