go-pkg/asynqmon
2
0
Fork 0
mirror of https://github.com/hibiken/asynqmon.git synced 2025-01-19 03:05:53 +08:00
Code Issues Projects Releases Wiki Activity

update BytesStringer => PayloadFormatter to pass taskType

Browse Source
This commit is contained in:
ajatprabha 2021-10-02 12:57:41 +05:30 committed by Ken Hibino
parent e635b73e6c
commit e569ad9186
6 changed files with 80 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
conversion_helpers.go
View File

@ -14,18 +14,18 @@ import (
// - conversion function from an external type to an internal type // - conversion function from an external type to an internal type
// **************************************************************************** // ****************************************************************************
// BytesStringer can be used to convert payload bytes to string to show in web ui. // PayloadFormatter can be used to convert payload bytes to string to show in web ui.
type BytesStringer interface { type PayloadFormatter interface {
String([]byte) string FormatPayload(taskType string, payload []byte) string
} }
type BytesStringerFunc func([]byte) string type PayloadFormatterFunc func(string, []byte) string
func (f BytesStringerFunc) String(b []byte) string { func (f PayloadFormatterFunc) FormatPayload(taskType string, payload []byte) string {
return f(b) return f(taskType, payload)
} }
var defaultBytesStringer = BytesStringerFunc(func(payload []byte) string { var defaultPayloadFormatter = PayloadFormatterFunc(func(_ string, payload []byte) string {
if !isPrintable(payload) { if !isPrintable(payload) {
return "non-printable bytes" return "non-printable bytes"
} }
@ -158,12 +158,12 @@ func formatTimeInRFC3339(t time.Time) string {
return t.Format(time.RFC3339) return t.Format(time.RFC3339)
} }
func toTaskInfo(info *asynq.TaskInfo, bs BytesStringer) *TaskInfo { func toTaskInfo(info *asynq.TaskInfo, pf PayloadFormatter) *TaskInfo {
return &TaskInfo{ return &TaskInfo{
ID: info.ID, ID: info.ID,
Queue: info.Queue, Queue: info.Queue,
Type: info.Type, Type: info.Type,
Payload: bs.String(info.Payload), Payload: pf.FormatPayload(info.Type, info.Payload),
State: info.State.String(), State: info.State.String(),
MaxRetry: info.MaxRetry, MaxRetry: info.MaxRetry,
Retried: info.Retried, Retried: info.Retried,
@ -202,11 +202,11 @@ type ActiveTask struct {
Deadline string `json:"deadline"` Deadline string `json:"deadline"`
} }
func toActiveTask(ti *asynq.TaskInfo, bs BytesStringer) *ActiveTask { func toActiveTask(ti *asynq.TaskInfo, pf PayloadFormatter) *ActiveTask {
base := &BaseTask{ base := &BaseTask{
ID: ti.ID, ID: ti.ID,
Type: ti.Type, Type: ti.Type,
Payload: bs.String(ti.Payload), Payload: pf.FormatPayload(ti.Type, ti.Payload),
Queue: ti.Queue, Queue: ti.Queue,
MaxRetry: ti.MaxRetry, MaxRetry: ti.MaxRetry,
Retried: ti.Retried, Retried: ti.Retried,
@ -215,10 +215,10 @@ func toActiveTask(ti *asynq.TaskInfo, bs BytesStringer) *ActiveTask {
return &ActiveTask{BaseTask: base} return &ActiveTask{BaseTask: base}
} }
func toActiveTasks(in []*asynq.TaskInfo, bs BytesStringer) []*ActiveTask { func toActiveTasks(in []*asynq.TaskInfo, pf PayloadFormatter) []*ActiveTask {
out := make([]*ActiveTask, len(in)) out := make([]*ActiveTask, len(in))
for i, ti := range in { for i, ti := range in {
out[i] = toActiveTask(ti, bs) out[i] = toActiveTask(ti, pf)
} }
return out return out
} }
@ -228,11 +228,11 @@ type PendingTask struct {
*BaseTask *BaseTask
} }
func toPendingTask(ti *asynq.TaskInfo, bs BytesStringer) *PendingTask { func toPendingTask(ti *asynq.TaskInfo, pf PayloadFormatter) *PendingTask {
base := &BaseTask{ base := &BaseTask{
ID: ti.ID, ID: ti.ID,
Type: ti.Type, Type: ti.Type,
Payload: bs.String(ti.Payload), Payload: pf.FormatPayload(ti.Type, ti.Payload),
Queue: ti.Queue, Queue: ti.Queue,
MaxRetry: ti.MaxRetry, MaxRetry: ti.MaxRetry,
Retried: ti.Retried, Retried: ti.Retried,
@ -243,10 +243,10 @@ func toPendingTask(ti *asynq.TaskInfo, bs BytesStringer) *PendingTask {
} }
} }
func toPendingTasks(in []*asynq.TaskInfo, bs BytesStringer) []*PendingTask { func toPendingTasks(in []*asynq.TaskInfo, pf PayloadFormatter) []*PendingTask {
out := make([]*PendingTask, len(in)) out := make([]*PendingTask, len(in))
for i, ti := range in { for i, ti := range in {
out[i] = toPendingTask(ti, bs) out[i] = toPendingTask(ti, pf)
} }
return out return out
} }
@ -256,11 +256,11 @@ type ScheduledTask struct {
NextProcessAt time.Time `json:"next_process_at"` NextProcessAt time.Time `json:"next_process_at"`
} }
func toScheduledTask(ti *asynq.TaskInfo, bs BytesStringer) *ScheduledTask { func toScheduledTask(ti *asynq.TaskInfo, pf PayloadFormatter) *ScheduledTask {
base := &BaseTask{ base := &BaseTask{
ID: ti.ID, ID: ti.ID,
Type: ti.Type, Type: ti.Type,
Payload: bs.String(ti.Payload), Payload: pf.FormatPayload(ti.Type, ti.Payload),
Queue: ti.Queue, Queue: ti.Queue,
MaxRetry: ti.MaxRetry, MaxRetry: ti.MaxRetry,
Retried: ti.Retried, Retried: ti.Retried,
@ -272,10 +272,10 @@ func toScheduledTask(ti *asynq.TaskInfo, bs BytesStringer) *ScheduledTask {
} }
} }
func toScheduledTasks(in []*asynq.TaskInfo, bs BytesStringer) []*ScheduledTask { func toScheduledTasks(in []*asynq.TaskInfo, pf PayloadFormatter) []*ScheduledTask {
out := make([]*ScheduledTask, len(in)) out := make([]*ScheduledTask, len(in))
for i, ti := range in { for i, ti := range in {
out[i] = toScheduledTask(ti, bs) out[i] = toScheduledTask(ti, pf)
} }
return out return out
} }
@ -285,11 +285,11 @@ type RetryTask struct {
NextProcessAt time.Time `json:"next_process_at"` NextProcessAt time.Time `json:"next_process_at"`
} }
func toRetryTask(ti *asynq.TaskInfo, bs BytesStringer) *RetryTask { func toRetryTask(ti *asynq.TaskInfo, pf PayloadFormatter) *RetryTask {
base := &BaseTask{ base := &BaseTask{
ID: ti.ID, ID: ti.ID,
Type: ti.Type, Type: ti.Type,
Payload: bs.String(ti.Payload), Payload: pf.FormatPayload(ti.Type, ti.Payload),
Queue: ti.Queue, Queue: ti.Queue,
MaxRetry: ti.MaxRetry, MaxRetry: ti.MaxRetry,
Retried: ti.Retried, Retried: ti.Retried,
@ -301,10 +301,10 @@ func toRetryTask(ti *asynq.TaskInfo, bs BytesStringer) *RetryTask {
} }
} }
func toRetryTasks(in []*asynq.TaskInfo, bs BytesStringer) []*RetryTask { func toRetryTasks(in []*asynq.TaskInfo, pf PayloadFormatter) []*RetryTask {
out := make([]*RetryTask, len(in)) out := make([]*RetryTask, len(in))
for i, ti := range in { for i, ti := range in {
out[i] = toRetryTask(ti, bs) out[i] = toRetryTask(ti, pf)
} }
return out return out
} }
@ -314,11 +314,11 @@ type ArchivedTask struct {
LastFailedAt time.Time `json:"last_failed_at"` LastFailedAt time.Time `json:"last_failed_at"`
} }
func toArchivedTask(ti *asynq.TaskInfo, bs BytesStringer) *ArchivedTask { func toArchivedTask(ti *asynq.TaskInfo, pf PayloadFormatter) *ArchivedTask {
base := &BaseTask{ base := &BaseTask{
ID: ti.ID, ID: ti.ID,
Type: ti.Type, Type: ti.Type,
Payload: bs.String(ti.Payload), Payload: pf.FormatPayload(ti.Type, ti.Payload),
Queue: ti.Queue, Queue: ti.Queue,
MaxRetry: ti.MaxRetry, MaxRetry: ti.MaxRetry,
Retried: ti.Retried, Retried: ti.Retried,
@ -330,10 +330,10 @@ func toArchivedTask(ti *asynq.TaskInfo, bs BytesStringer) *ArchivedTask {
} }
} }
func toArchivedTasks(in []*asynq.TaskInfo, bs BytesStringer) []*ArchivedTask { func toArchivedTasks(in []*asynq.TaskInfo, pf PayloadFormatter) []*ArchivedTask {
out := make([]*ArchivedTask, len(in)) out := make([]*ArchivedTask, len(in))
for i, ti := range in { for i, ti := range in {
out[i] = toArchivedTask(ti, bs) out[i] = toArchivedTask(ti, pf)
} }
return out return out
} }
@ -349,7 +349,7 @@ type SchedulerEntry struct {
PrevEnqueueAt string `json:"prev_enqueue_at,omitempty"` PrevEnqueueAt string `json:"prev_enqueue_at,omitempty"`
} }
func toSchedulerEntry(e *asynq.SchedulerEntry, bs BytesStringer) *SchedulerEntry { func toSchedulerEntry(e *asynq.SchedulerEntry, pf PayloadFormatter) *SchedulerEntry {
opts := make([]string, 0) // create a non-nil, empty slice to avoid null in json output opts := make([]string, 0) // create a non-nil, empty slice to avoid null in json output
for _, o := range e.Opts { for _, o := range e.Opts {
opts = append(opts, o.String()) opts = append(opts, o.String())
@ -362,17 +362,17 @@ func toSchedulerEntry(e *asynq.SchedulerEntry, bs BytesStringer) *SchedulerEntry
ID: e.ID, ID: e.ID,
Spec: e.Spec, Spec: e.Spec,
TaskType: e.Task.Type(), TaskType: e.Task.Type(),
TaskPayload: bs.String(e.Task.Payload()), TaskPayload: pf.FormatPayload(e.Task.Type(), e.Task.Payload()),
Opts: opts, Opts: opts,
NextEnqueueAt: e.Next.Format(time.RFC3339), NextEnqueueAt: e.Next.Format(time.RFC3339),
PrevEnqueueAt: prev, PrevEnqueueAt: prev,
} }
} }
func toSchedulerEntries(in []*asynq.SchedulerEntry, bs BytesStringer) []*SchedulerEntry { func toSchedulerEntries(in []*asynq.SchedulerEntry, pf PayloadFormatter) []*SchedulerEntry {
out := make([]*SchedulerEntry, len(in)) out := make([]*SchedulerEntry, len(in))
for i, e := range in { for i, e := range in {
out[i] = toSchedulerEntry(e, bs) out[i] = toSchedulerEntry(e, pf)
} }
return out return out
} }
@ -409,7 +409,7 @@ type ServerInfo struct {
ActiveWorkers []*WorkerInfo `json:"active_workers"` ActiveWorkers []*WorkerInfo `json:"active_workers"`
} }
func toServerInfo(info *asynq.ServerInfo, bs BytesStringer) *ServerInfo { func toServerInfo(info *asynq.ServerInfo, pf PayloadFormatter) *ServerInfo {
return &ServerInfo{ return &ServerInfo{
ID: info.ID, ID: info.ID,
Host: info.Host, Host: info.Host,
@ -419,40 +419,40 @@ func toServerInfo(info *asynq.ServerInfo, bs BytesStringer) *ServerInfo {
StrictPriority: info.StrictPriority, StrictPriority: info.StrictPriority,
Started: info.Started.Format(time.RFC3339), Started: info.Started.Format(time.RFC3339),
Status: info.Status, Status: info.Status,
ActiveWorkers: toWorkerInfoList(info.ActiveWorkers, bs), ActiveWorkers: toWorkerInfoList(info.ActiveWorkers, pf),
} }
} }
func toServerInfoList(in []*asynq.ServerInfo, bs BytesStringer) []*ServerInfo { func toServerInfoList(in []*asynq.ServerInfo, pf PayloadFormatter) []*ServerInfo {
out := make([]*ServerInfo, len(in)) out := make([]*ServerInfo, len(in))
for i, s := range in { for i, s := range in {
out[i] = toServerInfo(s, bs) out[i] = toServerInfo(s, pf)
} }
return out return out
} }
type WorkerInfo struct { type WorkerInfo struct {
TaskID string `json:"task_id"` TaskID string `json:"task_id"`
Queue string `json:"queue"` Queue string `json:"queue"`
TaskType string `json:"task_type"` TaskType string `json:"task_type"`
TakPayload string `json:"task_payload"` TaskPayload string `json:"task_payload"`
Started string `json:"start_time"` Started string `json:"start_time"`
} }
func toWorkerInfo(info *asynq.WorkerInfo, bs BytesStringer) *WorkerInfo { func toWorkerInfo(info *asynq.WorkerInfo, pf PayloadFormatter) *WorkerInfo {
return &WorkerInfo{ return &WorkerInfo{
TaskID: info.TaskID, TaskID: info.TaskID,
Queue: info.Queue, Queue: info.Queue,
TaskType: info.TaskType, TaskType: info.TaskType,
TakPayload: bs.String(info.TaskPayload), TaskPayload: pf.FormatPayload(info.TaskType, info.TaskPayload),
Started: info.Started.Format(time.RFC3339), Started: info.Started.Format(time.RFC3339),
} }
} }
func toWorkerInfoList(in []*asynq.WorkerInfo, bs BytesStringer) []*WorkerInfo { func toWorkerInfoList(in []*asynq.WorkerInfo, pf PayloadFormatter) []*WorkerInfo {
out := make([]*WorkerInfo, len(in)) out := make([]*WorkerInfo, len(in))
for i, w := range in { for i, w := range in {
out[i] = toWorkerInfo(w, bs) out[i] = toWorkerInfo(w, pf)
} }
return out return out
} }

26
handler.go
View File

@ -13,7 +13,7 @@ type HandlerOptions struct {
RedisClient redis.UniversalClient RedisClient redis.UniversalClient
Inspector *asynq.Inspector Inspector *asynq.Inspector
Middlewares []mux.MiddlewareFunc Middlewares []mux.MiddlewareFunc
BytesStringer BytesStringer PayloadFormatter PayloadFormatter
StaticContentHandler http.Handler StaticContentHandler http.Handler
} }
@ -21,9 +21,9 @@ func NewHandler(opts HandlerOptions) http.Handler {
router := mux.NewRouter() router := mux.NewRouter()
inspector := opts.Inspector inspector := opts.Inspector
var bs BytesStringer = defaultBytesStringer var pf PayloadFormatter = defaultPayloadFormatter
if opts.BytesStringer != nil { if opts.PayloadFormatter != nil {
bs = opts.BytesStringer pf = opts.PayloadFormatter
} }
for _, mf := range opts.Middlewares { for _, mf := range opts.Middlewares {
@ -32,7 +32,7 @@ func NewHandler(opts HandlerOptions) http.Handler {
api := router.PathPrefix("/api").Subrouter() api := router.PathPrefix("/api").Subrouter()
// Queue endpoints. // Queue endpoints.
api.HandleFunc("/queues", newListQueuesHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues", newListQueuesHandlerFunc(inspector)).Methods("GET")
api.HandleFunc("/queues/{qname}", newGetQueueHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}", newGetQueueHandlerFunc(inspector)).Methods("GET")
api.HandleFunc("/queues/{qname}", newDeleteQueueHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}", newDeleteQueueHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}:pause", newPauseQueueHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}:pause", newPauseQueueHandlerFunc(inspector)).Methods("POST")
@ -42,12 +42,12 @@ func NewHandler(opts HandlerOptions) http.Handler {
api.HandleFunc("/queue_stats", newListQueueStatsHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queue_stats", newListQueueStatsHandlerFunc(inspector)).Methods("GET")
// Task endpoints. // Task endpoints.
api.HandleFunc("/queues/{qname}/active_tasks", newListActiveTasksHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues/{qname}/active_tasks", newListActiveTasksHandlerFunc(inspector, pf)).Methods("GET")
api.HandleFunc("/queues/{qname}/active_tasks/{task_id}:cancel", newCancelActiveTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/active_tasks/{task_id}:cancel", newCancelActiveTaskHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/active_tasks:cancel_all", newCancelAllActiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/active_tasks:cancel_all", newCancelAllActiveTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/active_tasks:batch_cancel", newBatchCancelActiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/active_tasks:batch_cancel", newBatchCancelActiveTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/pending_tasks", newListPendingTasksHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues/{qname}/pending_tasks", newListPendingTasksHandlerFunc(inspector, pf)).Methods("GET")
api.HandleFunc("/queues/{qname}/pending_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/pending_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/pending_tasks:delete_all", newDeleteAllPendingTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/pending_tasks:delete_all", newDeleteAllPendingTasksHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/pending_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/pending_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST")
@ -55,7 +55,7 @@ func NewHandler(opts HandlerOptions) http.Handler {
api.HandleFunc("/queues/{qname}/pending_tasks:archive_all", newArchiveAllPendingTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/pending_tasks:archive_all", newArchiveAllPendingTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/pending_tasks:batch_archive", newBatchArchiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/pending_tasks:batch_archive", newBatchArchiveTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/scheduled_tasks", newListScheduledTasksHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues/{qname}/scheduled_tasks", newListScheduledTasksHandlerFunc(inspector, pf)).Methods("GET")
api.HandleFunc("/queues/{qname}/scheduled_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/scheduled_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/scheduled_tasks:delete_all", newDeleteAllScheduledTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/scheduled_tasks:delete_all", newDeleteAllScheduledTasksHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST")
@ -66,7 +66,7 @@ func NewHandler(opts HandlerOptions) http.Handler {
api.HandleFunc("/queues/{qname}/scheduled_tasks:archive_all", newArchiveAllScheduledTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:archive_all", newArchiveAllScheduledTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_archive", newBatchArchiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_archive", newBatchArchiveTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/retry_tasks", newListRetryTasksHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues/{qname}/retry_tasks", newListRetryTasksHandlerFunc(inspector, pf)).Methods("GET")
api.HandleFunc("/queues/{qname}/retry_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/retry_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/retry_tasks:delete_all", newDeleteAllRetryTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/retry_tasks:delete_all", newDeleteAllRetryTasksHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/retry_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST")
@ -77,7 +77,7 @@ func NewHandler(opts HandlerOptions) http.Handler {
api.HandleFunc("/queues/{qname}/retry_tasks:archive_all", newArchiveAllRetryTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:archive_all", newArchiveAllRetryTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/retry_tasks:batch_archive", newBatchArchiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:batch_archive", newBatchArchiveTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/archived_tasks", newListArchivedTasksHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues/{qname}/archived_tasks", newListArchivedTasksHandlerFunc(inspector, pf)).Methods("GET")
api.HandleFunc("/queues/{qname}/archived_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/archived_tasks/{task_id}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/archived_tasks:delete_all", newDeleteAllArchivedTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/archived_tasks:delete_all", newDeleteAllArchivedTasksHandlerFunc(inspector)).Methods("DELETE")
api.HandleFunc("/queues/{qname}/archived_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/archived_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST")
@ -85,13 +85,13 @@ func NewHandler(opts HandlerOptions) http.Handler {
api.HandleFunc("/queues/{qname}/archived_tasks:run_all", newRunAllArchivedTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/archived_tasks:run_all", newRunAllArchivedTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/archived_tasks:batch_run", newBatchRunTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/archived_tasks:batch_run", newBatchRunTasksHandlerFunc(inspector)).Methods("POST")
api.HandleFunc("/queues/{qname}/tasks/{task_id}", newGetTaskHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/queues/{qname}/tasks/{task_id}", newGetTaskHandlerFunc(inspector, pf)).Methods("GET")
// Servers endpoints. // Servers endpoints.
api.HandleFunc("/servers", newListServersHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/servers", newListServersHandlerFunc(inspector, pf)).Methods("GET")
// Scheduler Entry endpoints. // Scheduler Entry endpoints.
api.HandleFunc("/scheduler_entries", newListSchedulerEntriesHandlerFunc(inspector, bs)).Methods("GET") api.HandleFunc("/scheduler_entries", newListSchedulerEntriesHandlerFunc(inspector, pf)).Methods("GET")
api.HandleFunc("/scheduler_entries/{entry_id}/enqueue_events", newListSchedulerEnqueueEventsHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/scheduler_entries/{entry_id}/enqueue_events", newListSchedulerEnqueueEventsHandlerFunc(inspector)).Methods("GET")
// Redis info endpoint. // Redis info endpoint.

2
queue_handlers.go
View File

@ -15,7 +15,7 @@ import (
// - http.Handler(s) for queue related endpoints // - http.Handler(s) for queue related endpoints
// **************************************************************************** // ****************************************************************************
func newListQueuesHandlerFunc(inspector *asynq.Inspector, t BytesStringer) http.HandlerFunc { func newListQueuesHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
qnames, err := inspector.Queues() qnames, err := inspector.Queues()
if err != nil { if err != nil {

4
scheduler_entry_handlers.go
View File

@ -14,7 +14,7 @@ import (
// - http.Handler(s) for scheduler entry related endpoints // - http.Handler(s) for scheduler entry related endpoints
// **************************************************************************** // ****************************************************************************
func newListSchedulerEntriesHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListSchedulerEntriesHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
entries, err := inspector.SchedulerEntries() entries, err := inspector.SchedulerEntries()
if err != nil { if err != nil {
@ -26,7 +26,7 @@ func newListSchedulerEntriesHandlerFunc(inspector *asynq.Inspector, bs BytesStri
// avoid nil for the entries field in json output. // avoid nil for the entries field in json output.
payload["entries"] = make([]*SchedulerEntry, 0) payload["entries"] = make([]*SchedulerEntry, 0)
} else { } else {
payload["entries"] = toSchedulerEntries(entries, bs) payload["entries"] = toSchedulerEntries(entries, pf)
} }
if err := json.NewEncoder(w).Encode(payload); err != nil { if err := json.NewEncoder(w).Encode(payload); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)

4
server_handlers.go
View File

@ -16,7 +16,7 @@ type ListServersResponse struct {
Servers []*ServerInfo `json:"servers"` Servers []*ServerInfo `json:"servers"`
} }
func newListServersHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListServersHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
srvs, err := inspector.Servers() srvs, err := inspector.Servers()
if err != nil { if err != nil {
@ -24,7 +24,7 @@ func newListServersHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) htt
return return
} }
resp := ListServersResponse{ resp := ListServersResponse{
Servers: toServerInfoList(srvs, bs), Servers: toServerInfoList(srvs, pf),
} }
if err := json.NewEncoder(w).Encode(resp); err != nil { if err := json.NewEncoder(w).Encode(resp); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)

24
task_handlers.go
View File

@ -24,7 +24,7 @@ type ListActiveTasksResponse struct {
Stats *QueueStateSnapshot `json:"stats"` Stats *QueueStateSnapshot `json:"stats"`
} }
func newListActiveTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListActiveTasksHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
qname := vars["qname"] qname := vars["qname"]
@ -55,7 +55,7 @@ func newListActiveTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer)
} }
} }
} }
activeTasks := toActiveTasks(tasks, bs) activeTasks := toActiveTasks(tasks, pf)
for _, t := range activeTasks { for _, t := range activeTasks {
workerInfo, ok := m[t.ID] workerInfo, ok := m[t.ID]
if ok { if ok {
@ -156,7 +156,7 @@ func newBatchCancelActiveTasksHandlerFunc(inspector *asynq.Inspector) http.Handl
} }
} }
func newListPendingTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListPendingTasksHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
qname := vars["qname"] qname := vars["qname"]
@ -177,7 +177,7 @@ func newListPendingTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer
// avoid nil for the tasks field in json output. // avoid nil for the tasks field in json output.
payload["tasks"] = make([]*PendingTask, 0) payload["tasks"] = make([]*PendingTask, 0)
} else { } else {
payload["tasks"] = toPendingTasks(tasks, bs) payload["tasks"] = toPendingTasks(tasks, pf)
} }
payload["stats"] = toQueueStateSnapshot(qinfo) payload["stats"] = toQueueStateSnapshot(qinfo)
if err := json.NewEncoder(w).Encode(payload); err != nil { if err := json.NewEncoder(w).Encode(payload); err != nil {
@ -187,7 +187,7 @@ func newListPendingTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer
} }
} }
func newListScheduledTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListScheduledTasksHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
qname := vars["qname"] qname := vars["qname"]
@ -208,7 +208,7 @@ func newListScheduledTasksHandlerFunc(inspector *asynq.Inspector, bs BytesString
// avoid nil for the tasks field in json output. // avoid nil for the tasks field in json output.
payload["tasks"] = make([]*ScheduledTask, 0) payload["tasks"] = make([]*ScheduledTask, 0)
} else { } else {
payload["tasks"] = toScheduledTasks(tasks, bs) payload["tasks"] = toScheduledTasks(tasks, pf)
} }
payload["stats"] = toQueueStateSnapshot(qinfo) payload["stats"] = toQueueStateSnapshot(qinfo)
if err := json.NewEncoder(w).Encode(payload); err != nil { if err := json.NewEncoder(w).Encode(payload); err != nil {
@ -218,7 +218,7 @@ func newListScheduledTasksHandlerFunc(inspector *asynq.Inspector, bs BytesString
} }
} }
func newListRetryTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListRetryTasksHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
qname := vars["qname"] qname := vars["qname"]
@ -239,7 +239,7 @@ func newListRetryTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer)
// avoid nil for the tasks field in json output. // avoid nil for the tasks field in json output.
payload["tasks"] = make([]*RetryTask, 0) payload["tasks"] = make([]*RetryTask, 0)
} else { } else {
payload["tasks"] = toRetryTasks(tasks, bs) payload["tasks"] = toRetryTasks(tasks, pf)
} }
payload["stats"] = toQueueStateSnapshot(qinfo) payload["stats"] = toQueueStateSnapshot(qinfo)
if err := json.NewEncoder(w).Encode(payload); err != nil { if err := json.NewEncoder(w).Encode(payload); err != nil {
@ -249,7 +249,7 @@ func newListRetryTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer)
} }
} }
func newListArchivedTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newListArchivedTasksHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
qname := vars["qname"] qname := vars["qname"]
@ -270,7 +270,7 @@ func newListArchivedTasksHandlerFunc(inspector *asynq.Inspector, bs BytesStringe
// avoid nil for the tasks field in json output. // avoid nil for the tasks field in json output.
payload["tasks"] = make([]*ArchivedTask, 0) payload["tasks"] = make([]*ArchivedTask, 0)
} else { } else {
payload["tasks"] = toArchivedTasks(tasks, bs) payload["tasks"] = toArchivedTasks(tasks, pf)
} }
payload["stats"] = toQueueStateSnapshot(qinfo) payload["stats"] = toQueueStateSnapshot(qinfo)
if err := json.NewEncoder(w).Encode(payload); err != nil { if err := json.NewEncoder(w).Encode(payload); err != nil {
@ -627,7 +627,7 @@ func getPageOptions(r *http.Request) (pageSize, pageNum int) {
return pageSize, pageNum return pageSize, pageNum
} }
func newGetTaskHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.HandlerFunc { func newGetTaskHandlerFunc(inspector *asynq.Inspector, pf PayloadFormatter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
qname, taskid := vars["qname"], vars["task_id"] qname, taskid := vars["qname"], vars["task_id"]
@ -650,7 +650,7 @@ func newGetTaskHandlerFunc(inspector *asynq.Inspector, bs BytesStringer) http.Ha
return return
} }
if err := json.NewEncoder(w).Encode(toTaskInfo(info, bs)); err != nil { if err := json.NewEncoder(w).Encode(toTaskInfo(info, pf)); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
} }