package main import ( "time" "github.com/hibiken/asynq" ) // **************************************************************************** // This file defines: // - internal types with JSON struct tags // - conversion function from an external type to an internal type // **************************************************************************** type QueueStateSnapshot struct { // Name of the queue. Queue string `json:"queue"` // Total number of bytes the queue and its tasks require to be stored in redis. MemoryUsage int64 `json:"memory_usage_bytes"` // Total number of tasks in the queue. Size int `json:"size"` // Number of tasks in each state. Active int `json:"active"` Pending int `json:"pending"` Scheduled int `json:"scheduled"` Retry int `json:"retry"` Archived int `json:"archived"` // Total number of tasks processed during the given date. // The number includes both succeeded and failed tasks. Processed int `json:"processed"` // Breakdown of processed tasks. Succeeded int `json:"succeeded"` Failed int `json:"failed"` // Paused indicates whether the queue is paused. // If true, tasks in the queue will not be processed. Paused bool `json:"paused"` // Time when this snapshot was taken. Timestamp time.Time `json:"timestamp"` } func toQueueStateSnapshot(s *asynq.QueueStats) *QueueStateSnapshot { return &QueueStateSnapshot{ Queue: s.Queue, MemoryUsage: s.MemoryUsage, Size: s.Size, Active: s.Active, Pending: s.Pending, Scheduled: s.Scheduled, Retry: s.Retry, Archived: s.Archived, Processed: s.Processed, Succeeded: s.Processed - s.Failed, Failed: s.Failed, Paused: s.Paused, Timestamp: s.Timestamp, } } type DailyStats struct { Queue string `json:"queue"` Processed int `json:"processed"` Succeeded int `json:"succeeded"` Failed int `json:"failed"` Date string `json:"date"` } func toDailyStats(s *asynq.DailyStats) *DailyStats { return &DailyStats{ Queue: s.Queue, Processed: s.Processed, Succeeded: s.Processed - s.Failed, Failed: s.Failed, Date: s.Date.Format("2006-01-02"), } } func toDailyStatsList(in []*asynq.DailyStats) []*DailyStats { out := make([]*DailyStats, len(in)) for i, s := range in { out[i] = toDailyStats(s) } return out } type BaseTask struct { ID string `json:"id"` Type string `json:"type"` Payload asynq.Payload `json:"payload"` Queue string `json:"queue"` } type ActiveTask struct { *BaseTask // Started time indicates when a worker started working on ths task. // // Value is either time formatted in RFC3339 format, or "-" which indicates // a worker started working on the task only a few moments ago, and started time // data is not available. Started string `json:"start_time"` } func toActiveTask(t *asynq.ActiveTask) *ActiveTask { base := &BaseTask{ ID: t.ID, Type: t.Type, Payload: t.Payload, Queue: t.Queue, } return &ActiveTask{BaseTask: base} } func toActiveTasks(in []*asynq.ActiveTask) []*ActiveTask { out := make([]*ActiveTask, len(in)) for i, t := range in { out[i] = toActiveTask(t) } return out } type PendingTask struct { *BaseTask Key string `json:"key"` } func toPendingTask(t *asynq.PendingTask) *PendingTask { base := &BaseTask{ ID: t.ID, Type: t.Type, Payload: t.Payload, Queue: t.Queue, } return &PendingTask{ BaseTask: base, Key: t.Key(), } } func toPendingTasks(in []*asynq.PendingTask) []*PendingTask { out := make([]*PendingTask, len(in)) for i, t := range in { out[i] = toPendingTask(t) } return out } type ScheduledTask struct { *BaseTask Key string `json:"key"` NextProcessAt time.Time `json:"next_process_at"` } func toScheduledTask(t *asynq.ScheduledTask) *ScheduledTask { base := &BaseTask{ ID: t.ID, Type: t.Type, Payload: t.Payload, Queue: t.Queue, } return &ScheduledTask{ BaseTask: base, Key: t.Key(), NextProcessAt: t.NextProcessAt, } } func toScheduledTasks(in []*asynq.ScheduledTask) []*ScheduledTask { out := make([]*ScheduledTask, len(in)) for i, t := range in { out[i] = toScheduledTask(t) } return out } type RetryTask struct { *BaseTask Key string `json:"key"` NextProcessAt time.Time `json:"next_process_at"` MaxRetry int `json:"max_retry"` Retried int `json:"retried"` ErrorMsg string `json:"error_message"` } func toRetryTask(t *asynq.RetryTask) *RetryTask { base := &BaseTask{ ID: t.ID, Type: t.Type, Payload: t.Payload, Queue: t.Queue, } return &RetryTask{ BaseTask: base, Key: t.Key(), NextProcessAt: t.NextProcessAt, MaxRetry: t.MaxRetry, Retried: t.Retried, ErrorMsg: t.ErrorMsg, } } func toRetryTasks(in []*asynq.RetryTask) []*RetryTask { out := make([]*RetryTask, len(in)) for i, t := range in { out[i] = toRetryTask(t) } return out } type ArchivedTask struct { *BaseTask Key string `json:"key"` MaxRetry int `json:"max_retry"` Retried int `json:"retried"` ErrorMsg string `json:"error_message"` LastFailedAt time.Time `json:"last_failed_at"` } func toArchivedTask(t *asynq.ArchivedTask) *ArchivedTask { base := &BaseTask{ ID: t.ID, Type: t.Type, Payload: t.Payload, Queue: t.Queue, } return &ArchivedTask{ BaseTask: base, Key: t.Key(), MaxRetry: t.MaxRetry, Retried: t.Retried, ErrorMsg: t.ErrorMsg, LastFailedAt: t.LastFailedAt, } } func toArchivedTasks(in []*asynq.ArchivedTask) []*ArchivedTask { out := make([]*ArchivedTask, len(in)) for i, t := range in { out[i] = toArchivedTask(t) } return out } type SchedulerEntry struct { ID string `json:"id"` Spec string `json:"spec"` TaskType string `json:"task_type"` TaskPayload asynq.Payload `json:"task_payload"` Opts []string `json:"options"` NextEnqueueAt string `json:"next_enqueue_at"` // This field is omitted if there were no previous enqueue events. PrevEnqueueAt string `json:"prev_enqueue_at,omitempty"` } func toSchedulerEntry(e *asynq.SchedulerEntry) *SchedulerEntry { opts := make([]string, 0) // create a non-nil, empty slice to avoid null in json output for _, o := range e.Opts { opts = append(opts, o.String()) } prev := "" if !e.Prev.IsZero() { prev = e.Prev.Format(time.RFC3339) } return &SchedulerEntry{ ID: e.ID, Spec: e.Spec, TaskType: e.Task.Type, TaskPayload: e.Task.Payload, Opts: opts, NextEnqueueAt: e.Next.Format(time.RFC3339), PrevEnqueueAt: prev, } } func toSchedulerEntries(in []*asynq.SchedulerEntry) []*SchedulerEntry { out := make([]*SchedulerEntry, len(in)) for i, e := range in { out[i] = toSchedulerEntry(e) } return out } type SchedulerEnqueueEvent struct { TaskID string `json:"task_id"` EnqueuedAt string `json:"enqueued_at"` } func toSchedulerEnqueueEvent(e *asynq.SchedulerEnqueueEvent) *SchedulerEnqueueEvent { return &SchedulerEnqueueEvent{ TaskID: e.TaskID, EnqueuedAt: e.EnqueuedAt.Format(time.RFC3339), } } func toSchedulerEnqueueEvents(in []*asynq.SchedulerEnqueueEvent) []*SchedulerEnqueueEvent { out := make([]*SchedulerEnqueueEvent, len(in)) for i, e := range in { out[i] = toSchedulerEnqueueEvent(e) } return out } type ServerInfo struct { ID string `json:"id"` Host string `json:"host"` PID int `json:"pid"` Concurrency int `json:"concurrency"` Queues map[string]int `json:"queue_priorities"` StrictPriority bool `json:"strict_priority_enabled"` Started string `json:"start_time"` Status string `json:"status"` ActiveWorkers []*WorkerInfo `json:"active_workers"` } func toServerInfo(info *asynq.ServerInfo) *ServerInfo { return &ServerInfo{ ID: info.ID, Host: info.Host, PID: info.PID, Concurrency: info.Concurrency, Queues: info.Queues, StrictPriority: info.StrictPriority, Started: info.Started.Format(time.RFC3339), Status: info.Status, ActiveWorkers: toWorkerInfoList(info.ActiveWorkers), } } func toServerInfoList(in []*asynq.ServerInfo) []*ServerInfo { out := make([]*ServerInfo, len(in)) for i, s := range in { out[i] = toServerInfo(s) } return out } type WorkerInfo struct { Task *ActiveTask `json:"task"` Started string `json:"start_time"` } func toWorkerInfo(info *asynq.WorkerInfo) *WorkerInfo { return &WorkerInfo{ Task: toActiveTask(info.Task), Started: info.Started.Format(time.RFC3339), } } func toWorkerInfoList(in []*asynq.WorkerInfo) []*WorkerInfo { out := make([]*WorkerInfo, len(in)) for i, w := range in { out[i] = toWorkerInfo(w) } return out }