Feature/bus job chain, Allows adding multiple tasks to be executed sequentially.#1082
Open
seth-shi wants to merge 5 commits intohibiken:masterfrom
Open
Feature/bus job chain, Allows adding multiple tasks to be executed sequentially.#1082seth-shi wants to merge 5 commits intohibiken:masterfrom
seth-shi wants to merge 5 commits intohibiken:masterfrom
Conversation
added 3 commits
October 25, 2025 13:56
…andling - Updated task timeout comment for clarity in `chain_test.go`. - Improved task verification by replacing `LPop` with `broker.Dequeue` for better reliability in multiple test cases. - Increased wait time for task completion in `TestChainIntegration_EndToEnd` to ensure reliability. - Added checks for task options and deadlines in chain tasks to ensure correct behavior.
seth-shi
changed the title
added 2 commits
October 26, 2025 10:01
- Introduced `TestChainWithDeadlineOption` to verify that the Deadline option is preserved during chain execution. - Added `TestChainWithProcessAtOption` to ensure the ProcessAt option is correctly handled and scheduled. - Updated comments in `chain.go` to clarify JSON marshaling behavior for time-related options.
Author
|
Run these three commands. Note that these three commands can be run simultaneously.
this is a example code // Copyright 2020 Kentaro Hibino. All rights reserved.
// Use of this source code is governed by a MIT license
// that can be found in the LICENSE file.
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"log"
"os"
"time"
"github.com/hibiken/asynq"
)
const (
TypeTask1 = "task:1"
TypeTask2 = "task:2"
TypeTask3 = "task:3"
)
type TaskPayload struct {
Name string `json:"name"`
}
var (
redisAddr = "localhost:6700"
)
func main() {
if len(os.Args) < 2 {
os.Exit(1)
}
mode := os.Args[1]
switch mode {
case "producer":
runProducer()
case "consumer1":
runConsumer("queue1", "Consumer1")
case "consumer2":
runConsumer("queue2", "Consumer2")
default:
fmt.Printf("Unknown mode: %s\n\n", mode)
os.Exit(1)
}
}
// ============================================
// Producer Mode
// ============================================
func runProducer() {
fs := flag.NewFlagSet("producer", flag.ExitOnError)
retention := fs.Duration("retention", 24*time.Hour, "Completed task retention time (e.g., 1h, 24h)")
fs.Parse(os.Args[2:])
// Create Redis client connection
client := asynq.NewClient(asynq.RedisClientOpt{
Addr: redisAddr,
})
defer client.Close()
// Create payloads for three tasks
payload1, _ := json.Marshal(TaskPayload{Name: "Task 1"})
payload2, _ := json.Marshal(TaskPayload{Name: "Task 2"})
payload3, _ := json.Marshal(TaskPayload{Name: "Task 3"})
// Create three tasks with different queues
// Use Retention to keep completed tasks visible in asynqmon
var opts []asynq.Option
if *retention > 0 {
opts = append(opts, asynq.Retention(*retention))
}
task1 := asynq.NewTask(TypeTask1, payload1,
append(opts, asynq.Queue("queue1"))...)
task2 := asynq.NewTask(TypeTask2, payload2,
append(opts, asynq.Queue("queue2"))...)
task3 := asynq.NewTask(TypeTask3, payload3,
append(opts, asynq.Queue("queue1"))...)
// Create chain task
chainTask := asynq.NewChainTask(task1, task2, task3)
// Enqueue chain task
info, err := client.Enqueue(chainTask)
if err != nil {
log.Fatalf("Failed to enqueue chain task: %v", err)
}
fmt.Println("========================================")
fmt.Println("✓ Chain task sent successfully!")
fmt.Println("========================================")
fmt.Printf(" Task ID: %s\n", info.ID)
fmt.Printf(" Initial Queue: %s\n", info.Queue)
fmt.Printf(" Task Chain:\n")
fmt.Printf(" 1. %s → queue1\n", TypeTask1)
fmt.Printf(" 2. %s → queue2\n", TypeTask2)
fmt.Printf(" 3. %s → queue1\n", TypeTask3)
if *retention > 0 {
fmt.Printf(" Retention: %s\n", *retention)
fmt.Println(" Tip: Completed tasks will appear in asynqmon's Completed tab")
} else {
fmt.Println(" Tip: Completed tasks will be deleted immediately")
fmt.Println(" Use --retention=24h to keep completed tasks")
}
fmt.Println("========================================")
}
// ============================================
// Consumer Mode
// ============================================
func runConsumer(queueName, consumerName string) {
fs := flag.NewFlagSet("consumer", flag.ExitOnError)
concurrency := fs.Int("concurrency", 5, "Number of concurrent workers")
fs.Parse(os.Args[2:])
// Create Redis server configuration
srv := asynq.NewServer(
asynq.RedisClientOpt{Addr: redisAddr},
asynq.Config{
Concurrency: *concurrency,
Queues: map[string]int{
queueName: 10,
},
},
)
// Create task handler
mux := asynq.NewServeMux()
// Register handlers for all task types
mux.HandleFunc(TypeTask1, makeHandler(queueName, consumerName))
mux.HandleFunc(TypeTask2, makeHandler(queueName, consumerName))
mux.HandleFunc(TypeTask3, makeHandler(queueName, consumerName))
fmt.Println("========================================")
fmt.Printf("%s started [Queue: %s]\n", consumerName, queueName)
fmt.Println("========================================")
fmt.Printf("Redis: %s\n", redisAddr)
fmt.Printf("Concurrency: %d\n", *concurrency)
fmt.Println("Waiting for tasks...")
fmt.Println("Press Ctrl+C to stop")
fmt.Println("========================================")
// Start server to process tasks
if err := srv.Run(mux); err != nil {
log.Fatalf("Failed to start server: %v", err)
}
}
// makeHandler creates a handler function for specific queue
func makeHandler(queueName, consumerName string) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error {
var payload TaskPayload
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("failed to unmarshal payload: %w", err)
}
// Output task start info
fmt.Printf("\n[%s][%s][%s] Started: %s (Type: %s)\n",
consumerName,
queueName,
time.Now().Format("15:04:05"),
payload.Name,
t.Type())
// Sleep for 3 seconds
for i := 1; i <= 3; i++ {
// Check if context is cancelled
select {
case <-ctx.Done():
fmt.Printf("[%s][%s][%s] ⚠ Cancelled: %s\n",
consumerName,
queueName,
time.Now().Format("15:04:05"),
payload.Name)
return ctx.Err()
default:
}
time.Sleep(1 * time.Second)
fmt.Printf(" [%s][%s][%s] %s processing... (%d/3s)\n",
consumerName,
queueName,
time.Now().Format("15:04:05"),
payload.Name,
i)
}
// Output task completion info
fmt.Printf("[%s][%s][%s] ✓ Completed: %s\n",
consumerName,
queueName,
time.Now().Format("15:04:05"),
payload.Name)
fmt.Println("----------------------------------------")
return nil
}
} |
|
I need this! |
| return fmt.Errorf("asynq: server cannot run with nil handler") | ||
| } | ||
|
|
||
| // Automatically inject Chain Middleware |
Collaborator
There was a problem hiding this comment.
Not sure about this tbh. Might be worth considering this a candidate for the x folder.
Collaborator
There was a problem hiding this comment.
We don't inject a middleware anywhere in the core. I would suggest leaving this up to the user. You could move this into the x/chain folder.
Author
There was a problem hiding this comment.
i have some question, can you give me some tips.
- chain dependence root asynq module, but i will register chain middleware in asynq.Start
- Task.w and Task.opts don't have export, i cannot use it in
xfolder (should i export these attribute or anther way)
Collaborator
There was a problem hiding this comment.
Yeah that's a problem as well....Let me think about this.
kamikazechaser
added
pr-feat
pr-alternative-available
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #1082 +/- ##
==========================================
+ Coverage 67.13% 69.01% +1.87%
==========================================
Files 29 30 +1
Lines 4300 5102 +802
==========================================
+ Hits 2887 3521 +634
- Misses 1135 1294 +159
- Partials 278 287 +9 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add Chain Task Support
Summary
Add
NewChainTask()for sequential task execution. Tasks run in order, chain stops on failure.Usage
Features
Files
chain.go(256 lines) - Core implementationchain_test.go(1,135 lines) - Test suiteserver.go- Register middleware