worker: added graceful shutdown functionality + client factory func

worker/_example/main.go

@@ -23,7 +23,7 @@ func main() {
 	}
 
 	// Init asynq client
-	asynqClient := asynq.NewClient(redisConnOpt)
+	asynqClient, err := asynq.NewClient(redisConnOpt)
 	defer asynqClient.Close()
 
 	// Create a context with a timeout and set the Server's context
@@ -38,6 +38,8 @@ func main() {
 		redisConnOpt, logger,
 		worker.WithSchedulerLocation("UTC"), // options are not required
 	)
+	defer schedulerServer.Shutdown()
+
 	// Init scheduler handlers
 	testScheduler := NewScheduler(nil)
 	// Run the scheduler
@@ -51,6 +53,8 @@ func main() {
 		redisConnOpt, logger,
 		worker.WithQueueName("default"), // options are not required
 	)
+	defer queueServer.Shutdown()
+
 	// Init worker handlers
 	testWorker := NewWorker(nil)
 	// Run the worker

worker/client.go

@@ -0,0 +1,19 @@
+package worker
+
+import (
+	"fmt"
+
+	"github.com/hibiken/asynq"
+)
+
+// NewClient creates a new asynq client from the given redis client instance.
+func NewClient(redisConnStr string) (*asynq.Client, asynq.RedisConnOpt, error) {
+	// Redis connect options for asynq client
+	redisConnOpt, err := asynq.ParseRedisURI(redisConnStr)
+	if err != nil {
+		return nil, nil, fmt.Errorf("worket.NewClient: failed to parse redis connection string: %w", err)
+	}
+
+	// Init asynq client
+	return asynq.NewClient(redisConnOpt), redisConnOpt, nil
+}

worker/queue.go

@@ -85,3 +85,10 @@ func registerQueueHandlers(handlers ...taskHandler) *asynq.ServeMux {
 
 	return mux
 }
+
+// Shutdown gracefully shuts down the queue server by waiting for all
+// in-flight tasks to finish processing before shutdown.
+func (srv *QueueServer) Shutdown() {
+	srv.Server.Stop()
+	srv.Server.Shutdown()
+}

worker/scheduler.go

@@ -66,3 +66,9 @@ func (srv *SchedulerServer) Run(handlers ...schedulerHandler) func() error {
 		return srv.Scheduler.Run()
 	}
 }
+
+// Shutdown gracefully shuts down the scheduler server by waiting for all
+// pending tasks to be processed.
+func (srv *SchedulerServer) Shutdown() {
+	srv.Scheduler.Shutdown()
+}