gocelery

connectOpions, err := redis.ParseURL("redis://")
if err != nil {
	panic(err)
}

client := redis.NewClient(connectOpions)

// initialize celery client
cli, _ := NewCeleryClient(
	NewRedisBroker(client),
	&RedisCeleryBackend{UniversalClient: client},
	1,
)

// prepare arguments
taskName := "worker.add"
argA := rand.Intn(10)
argB := rand.Intn(10)

task := GetTaskMessage(taskName)
task.Args = append(task.Args, argA, argB)

// run task
asyncResult, err := cli.Delay(context.Background(), task)
if err != nil {
	panic(err)
}

// get results from backend with timeout
res, err := asyncResult.Get(context.Background(), 10*time.Second)
if err != nil {
	panic(err)
}

log.Printf("result: %+v of type %+v", res, reflect.TypeOf(res))

Output:

connectOpions, err := redis.ParseURL("redis://")
if err != nil {
	panic(err)
}

client := redis.NewClient(connectOpions)

// initialize celery client
cli, _ := NewCeleryClient(
	NewRedisBroker(client),
	&RedisCeleryBackend{UniversalClient: client},
	1,
)

// prepare arguments
taskName := "worker.add"
argA := rand.Intn(10)
argB := rand.Intn(10)

// run task
asyncResult, err := cli.DelayKwargs(
	context.Background(),
	taskName,
	map[string]interface{}{
		"a": argA,
		"b": argB,
	},
)
if err != nil {
	panic(err)
}

// get results from backend with timeout
res, err := asyncResult.Get(context.Background(), 10*time.Second)
if err != nil {
	panic(err)
}

log.Printf("result: %+v of type %+v", res, reflect.TypeOf(res))

Output:

connectOpions, err := redis.ParseURL("redis://")
if err != nil {
	panic(err)
}

client := redis.NewClient(connectOpions)

// initialize celery client
cli, _ := NewCeleryClient(
	NewRedisBroker(client),
	&RedisCeleryBackend{UniversalClient: client},
	5, // number of workers
)

// task
add := func(a, b int) int {
	return a + b
}

// register task
cli.Register("add", add)

// start workers (non-blocking call)
cli.StartWorker()

// wait for client request
time.Sleep(10 * time.Second)

// stop workers gracefully (blocking call)
cli.StopWorker()

Output:

connectOpions, err := redis.ParseURL("redis://")
if err != nil {
	panic(err)
}

client := redis.NewClient(connectOpions)

// initialize celery client
cli, _ := NewCeleryClient(
	NewRedisBroker(client),
	&RedisCeleryBackend{UniversalClient: client},
	1,
)

// task
add := func(a, b int) int {
	return a + b
}

// register task
cli.Register("add", add)

// context with cancelFunc to handle exit gracefully
ctx, cancel := context.WithCancel(context.Background())

// start workers (non-blocking call)
cli.StartWorkerWithContext(ctx)

// wait for client request
time.Sleep(10 * time.Second)

// stop workers by cancelling context
cancel()

// optional: wait for all workers to terminate
cli.WaitForStopWorker()

Output:

package main

import (
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// exampleAddTask is integer addition task
// with named arguments
type exampleAddTask struct {
	a int
	b int
}

func (a *exampleAddTask) ParseKwargs(kwargs map[string]interface{}) error {
	kwargA, ok := kwargs["a"]
	if !ok {
		return fmt.Errorf("undefined kwarg a")
	}
	kwargAFloat, ok := kwargA.(float64)
	if !ok {
		return fmt.Errorf("malformed kwarg a")
	}
	a.a = int(kwargAFloat)
	kwargB, ok := kwargs["b"]
	if !ok {
		return fmt.Errorf("undefined kwarg b")
	}
	kwargBFloat, ok := kwargB.(float64)
	if !ok {
		return fmt.Errorf("malformed kwarg b")
	}
	a.b = int(kwargBFloat)
	return nil
}

func (a *exampleAddTask) RunTask() (interface{}, error) {
	result := a.a + a.b
	return result, nil
}

func main() {
	connectOpions, err := redis.ParseURL("redis://")
	if err != nil {
		panic(err)
	}

	client := redis.NewClient(connectOpions)

	// initialize celery client
	cli, _ := NewCeleryClient(
		NewRedisBroker(client),
		&RedisCeleryBackend{UniversalClient: client},
		5, // number of workers
	)

	// register task
	cli.Register("add", &exampleAddTask{})

	// start workers (non-blocking call)
	cli.StartWorker()

	// wait for client request
	time.Sleep(10 * time.Second)

	// stop workers gracefully (blocking call)
	cli.StopWorker()
}

Output: