[GH-ISSUE #218] [FEATURE REQUEST] return result to middleware

kerem commented

2026-03-07 22:05:34 +03:00

Owner

Originally created by @ChristianYeah on GitHub (Dec 17, 2020).
Original GitHub issue: https://github.com/hibiken/asynq/issues/218

Originally assigned to: @hibiken on GitHub.

in some circumstances, a chained job task is required.

I prefer not to write the logic in the current task to re-queue the post-process task

however, the payload is not mutable and only errors will return by the task handler, the middleware is not able to catch any result directly(except the Redis, etc.)

I prefer to consider this as a discussion rather than a feature request, I'm not pretty professional in the system's structure design.

thank you

Originally created by @ChristianYeah on GitHub (Dec 17, 2020). Original GitHub issue: https://github.com/hibiken/asynq/issues/218 Originally assigned to: @hibiken on GitHub. in some circumstances, a chained job task is required. I prefer not to write the logic in the current task to re-queue the post-process task however, the payload is not mutable and only errors will return by the task handler, the middleware is not able to catch any result directly(except the Redis, etc.) I prefer to consider this as a discussion rather than a feature request, I'm not pretty professional in the system's structure design. thank you

kerem

2026-03-07 22:05:34 +03:00

closed this issue
added the
enhancement
label

kerem commented

2026-03-07 22:05:36 +03:00

Author

Owner

@hibiken commented on GitHub (Dec 17, 2020):

Hi @ChristianYeah! Thank you for starting this discussion.

I believe this is a common need, to create some kind of a workflow of sort.
Examples:

TaskB should be processed only after TaskA has been successfully processed.
TaskD should be processed only after TaskA, TaskB, TaskC have been successfully processed.

Do you have ideas on how the API should look like from the user of the library point of view?

@hibiken commented on GitHub (Dec 17, 2020): Hi @ChristianYeah! Thank you for starting this discussion. I believe this is a common need, to create some kind of a workflow of sort. Examples: - TaskB should be processed only after TaskA has been successfully processed. - TaskD should be processed only after TaskA, TaskB, TaskC have been successfully processed. --- Do you have ideas on how the API should look like from the user of the library point of view?

kerem commented

2026-03-07 22:05:36 +03:00

Author

Owner

@sujit-baniya commented on GitHub (Dec 21, 2020):

@hibiken There's a library for workflow: https://github.com/rdleal/go-async

Maybe the Auto part is something you might consider.

package main

import (
        "context"
        "encoding/json"
        "fmt"
        "log"
        "strings"

        "github.com/rdleal/go-async/async"
)

func main() { 
	type msg struct {
		Greet, Name string
	}

	const jsonStr = `{"greet": "hello", "name" : "gopher"}`
	fnc, err := async.Auto(context.Background(), async.FuncMap{
		"reader":  strings.NewReader,
		"decoder": async.DependsOn("reader").To(json.NewDecoder),
		"parser": async.DependsOn("decoder").To(func(d *json.Decoder) (msg, error) {
			var m msg
			err := d.Decode(&m)
			return m, err
		}),
		"greet": async.DependsOn("parser").To(func(m msg) string {
			return strings.Title(m.Greet)
		}),
		"name": async.DependsOn("parser").To(func(m msg) string {
			return strings.Title(m.Name)
		}),
		"greeter": async.DependsOn("greet", "name").To(func(greet, name string) string {
			return fmt.Sprintf("%s, %s!", greet, name)
		}),
	}, async.WithArgs(jsonStr))
	if err != nil {
		log.Fatal(err)
	}

	for fn := range fnc {
		res, err := fn.Returned()
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println(res[0]) // Hello, Gopher!
	}
}

@sujit-baniya commented on GitHub (Dec 21, 2020): @hibiken There's a library for workflow: https://github.com/rdleal/go-async Maybe the `Auto` part is something you might consider. ```go package main import ( "context" "encoding/json" "fmt" "log" "strings" "github.com/rdleal/go-async/async" ) func main() { type msg struct { Greet, Name string } const jsonStr = `{"greet": "hello", "name" : "gopher"}` fnc, err := async.Auto(context.Background(), async.FuncMap{ "reader": strings.NewReader, "decoder": async.DependsOn("reader").To(json.NewDecoder), "parser": async.DependsOn("decoder").To(func(d *json.Decoder) (msg, error) { var m msg err := d.Decode(&m) return m, err }), "greet": async.DependsOn("parser").To(func(m msg) string { return strings.Title(m.Greet) }), "name": async.DependsOn("parser").To(func(m msg) string { return strings.Title(m.Name) }), "greeter": async.DependsOn("greet", "name").To(func(greet, name string) string { return fmt.Sprintf("%s, %s!", greet, name) }), }, async.WithArgs(jsonStr)) if err != nil { log.Fatal(err) } for fn := range fnc { res, err := fn.Returned() if err != nil { log.Fatal(err) } fmt.Println(res[0]) // Hello, Gopher! } } ```

kerem commented

2026-03-07 22:05:36 +03:00

Author

Owner

@sujit-baniya commented on GitHub (Dec 21, 2020):

Also in my view, for this library we need

to have some kind of event handling based on task state.
pass result of one task to another task

@sujit-baniya commented on GitHub (Dec 21, 2020): Also in my view, for this library we need - to have some kind of event handling based on task state. - pass result of one task to another task

kerem commented

2026-03-07 22:05:36 +03:00

Author

Owner

@ChristianYeah commented on GitHub (Dec 21, 2020):

Hi @ChristianYeah! Thank you for starting this discussion.

I believe this is a common need, to create some kind of a workflow of sort.
Examples:

TaskB should be processed only after TaskA has been successfully processed.

TaskD should be processed only after TaskA, TaskB, TaskC have been successfully processed.

Do you have ideas on how the API should look like from the user of the library point of view?

I think the simplest way is to allow the worker&task handler redirects its result to middleware or post-process middleware.
So developers could define the workflow in the payload rather than in golang code(rebuild & re-deploy are needed). users just need to ensure it's a non-cyclic graph or a DAG.

@ChristianYeah commented on GitHub (Dec 21, 2020): > Hi @ChristianYeah! Thank you for starting this discussion. > > I believe this is a common need, to create some kind of a workflow of sort. > Examples: > > * TaskB should be processed only after TaskA has been successfully processed. > * TaskD should be processed only after TaskA, TaskB, TaskC have been successfully processed. > > Do you have ideas on how the API should look like from the user of the library point of view? I think the simplest way is to allow the worker&task handler redirects its result to middleware or post-process middleware. So developers could define the workflow in the payload rather than in golang code(rebuild & re-deploy are needed). users just need to ensure it's a non-cyclic graph or a DAG.

kerem commented

2026-03-07 22:05:36 +03:00

Author

Owner

@hibiken commented on GitHub (Dec 22, 2020):

@ChristianYeah Do you think storing a result in the context would work? Using WithValue from context package.

I'm interested in building a feature for this in a longer term. I'll study how Celery and Sidekiq is handling this type of workflow.

@hibiken commented on GitHub (Dec 22, 2020): @ChristianYeah Do you think storing a result in the context would work? Using [`WithValue`](https://golang.org/pkg/context/#WithValue) from context package. I'm interested in building a feature for this in a longer term. I'll study how [Celery](https://docs.celeryproject.org/en/stable/userguide/canvas.html) and [Sidekiq](https://github.com/mperham/sidekiq/wiki/Batches) is handling this type of workflow.

kerem commented

2026-03-07 22:05:37 +03:00

Author

Owner

@ChristianYeah commented on GitHub (Dec 23, 2020):

@hibiken I tried to override the context via WithValue function in task's ProcessTask function. however, I'm not able to get the value in the context in the middleware.

@ChristianYeah commented on GitHub (Dec 23, 2020): @hibiken I tried to override the context via ```WithValue``` function in task's ```ProcessTask``` function. however, I'm not able to get the value in the context in the middleware.

kerem commented

2026-03-07 22:05:37 +03:00

Author

Owner

@hibiken commented on GitHub (Dec 23, 2020):

Yes, you are right, it's not possible to store a result in context and make it available to middleware since context is immutable.
This may require API change in Handler interface to return a result as well as an error. I'll work in the next iteration after I wrap up my current work on web ui.

In the meantime, maybe you could store the result in redis and access it in middleware?

@hibiken commented on GitHub (Dec 23, 2020): Yes, you are right, it's not possible to store a result in context and make it available to middleware since context is immutable. This may require API change in `Handler` interface to return a result as well as an error. I'll work in the next iteration after I wrap up my current work on web ui. In the meantime, maybe you could store the result in redis and access it in middleware?

kerem commented

2026-03-07 22:05:37 +03:00

Author

Owner

@hibiken commented on GitHub (Feb 15, 2021):

I'll move the discussion to #244 and close this one.

@hibiken commented on GitHub (Feb 15, 2021): I'll move the discussion to #244 and close this one.

kerem referenced this issue

2026-03-07 22:14:19 +03:00

[PR #1091] 支持配置LeaseDuration、HeartbeatInterval #2025

kerem referenced this issue

2026-03-15 21:16:31 +03:00