Initial commit

task_handlers.go

@@ -0,0 +1,169 @@
+package main
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+
+	"github.com/gorilla/mux"
+	"github.com/hibiken/asynq"
+)
+
+func newListActiveTasksHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		qname := vars["qname"]
+		pageSize, pageNum := getPageOptions(r)
+		tasks, err := inspector.ListActiveTasks(
+			qname, asynq.PageSize(pageSize), asynq.Page(pageNum))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		stats, err := inspector.CurrentStats(qname)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		payload := make(map[string]interface{})
+		if len(tasks) == 0 {
+			// avoid nil for the tasks field in json output.
+			payload["tasks"] = make([]*asynq.ActiveTask, 0)
+		} else {
+			payload["tasks"] = toActiveTasks(tasks)
+		}
+		payload["stats"] = toQueueStateSnapshot(stats)
+		json.NewEncoder(w).Encode(payload)
+	}
+}
+
+func newListPendingTasksHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		qname := vars["qname"]
+		pageSize, pageNum := getPageOptions(r)
+		tasks, err := inspector.ListPendingTasks(
+			qname, asynq.PageSize(pageSize), asynq.Page(pageNum))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		stats, err := inspector.CurrentStats(qname)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		payload := make(map[string]interface{})
+		if len(tasks) == 0 {
+			// avoid nil for the tasks field in json output.
+			payload["tasks"] = make([]*asynq.PendingTask, 0)
+		} else {
+			payload["tasks"] = toPendingTasks(tasks)
+		}
+		payload["stats"] = toQueueStateSnapshot(stats)
+		json.NewEncoder(w).Encode(payload)
+	}
+}
+
+func newListScheduledTasksHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		qname := vars["qname"]
+		pageSize, pageNum := getPageOptions(r)
+		tasks, err := inspector.ListScheduledTasks(
+			qname, asynq.PageSize(pageSize), asynq.Page(pageNum))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		stats, err := inspector.CurrentStats(qname)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		payload := make(map[string]interface{})
+		if len(tasks) == 0 {
+			// avoid nil for the tasks field in json output.
+			payload["tasks"] = make([]*asynq.ScheduledTask, 0)
+		} else {
+			payload["tasks"] = toScheduledTasks(tasks)
+		}
+		payload["stats"] = toQueueStateSnapshot(stats)
+		json.NewEncoder(w).Encode(payload)
+	}
+}
+
+func newListRetryTasksHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		qname := vars["qname"]
+		pageSize, pageNum := getPageOptions(r)
+		tasks, err := inspector.ListRetryTasks(
+			qname, asynq.PageSize(pageSize), asynq.Page(pageNum))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		stats, err := inspector.CurrentStats(qname)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		payload := make(map[string]interface{})
+		if len(tasks) == 0 {
+			// avoid nil for the tasks field in json output.
+			payload["tasks"] = make([]*asynq.RetryTask, 0)
+		} else {
+			payload["tasks"] = toRetryTasks(tasks)
+		}
+		payload["stats"] = toQueueStateSnapshot(stats)
+		json.NewEncoder(w).Encode(payload)
+	}
+}
+
+func newListDeadTasksHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		qname := vars["qname"]
+		pageSize, pageNum := getPageOptions(r)
+		tasks, err := inspector.ListDeadTasks(
+			qname, asynq.PageSize(pageSize), asynq.Page(pageNum))
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		stats, err := inspector.CurrentStats(qname)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		payload := make(map[string]interface{})
+		if len(tasks) == 0 {
+			// avoid nil for the tasks field in json output.
+			payload["tasks"] = make([]*asynq.DeadTask, 0)
+		} else {
+			payload["tasks"] = toDeadTasks(tasks)
+		}
+		payload["stats"] = toQueueStateSnapshot(stats)
+		json.NewEncoder(w).Encode(payload)
+	}
+}
+
+// getPageOptions read page size and number from the request url if set,
+// otherwise it returns the default value.
+func getPageOptions(r *http.Request) (pageSize, pageNum int) {
+	pageSize = 20 // default page size
+	pageNum = 1   // default page num
+	q := r.URL.Query()
+	if s := q.Get("size"); s != "" {
+		if n, err := strconv.Atoi(s); err == nil {
+			pageSize = n
+		}
+	}
+	if s := q.Get("page"); s != "" {
+		if n, err := strconv.Atoi(s); err == nil {
+			pageNum = n
+		}
+	}
+	return pageSize, pageNum
+}