goworker

To create a worker, define a function with the signature func(string, ...interface{}) error. Register this function using goworker.Register("ClassName", functionName) and start the worker loop by calling goworker.Work().

package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func myFunc(queue string, args ...interface{}) error {
	fmt.Printf("From %s, %v\n", queue, args)
	return nil
}

func init() {
	goworker.Register("MyClass", myFunc)
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}

To install the goworker package, use the go get command and then import it into your Go worker application.

go get github.com/benmanns/goworker

import "github.com/benmanns/goworker"

goworker does not guarantee job safety during process shutdown.

• Idempotency: Workers must be idempotent and tolerant to job loss.
• KILL/System Failure: If the process is killed with KILL or a system failure, the job currently in the poller's buffer may be lost.
• Heroku/TERM signals: On platforms like Heroku, you have 10 seconds after a TERM signal before a KILL is sent. Jobs must finish within this window or they may be lost.
• Recovery: Lost jobs can sometimes be recovered from Redis under the key resque:worker:<hostname>:<process-id>-<worker-id>:<queues>, which contains a JSON object with queue, run_at, and payload. However, manual processing is required and there is no guarantee the job hasn't already finished.

Worker functions receive arguments as a slice of interface{}. Because Resque/goworker often uses JSON for payloads, you should use Go type assertions (specifically with json.Number for numeric values) to convert arguments into usable types.

// Expecting (int, string, float64)
func myFunc(queue, args ...interface{}) error {
	idNum, ok := args[0].(json.Number)
	if !ok {
		return errorInvalidParam
	}
	id, err := idNum.Int64()
	if err != nil {
		return errorInvalidParam
	}
	name, ok := args[1].(string)
	if !ok {
		return errorInvalidParam
	}
	weightNum, ok := args[2].(json.Number)
	if !ok {
		return errorInvalidParam
	}
	weight, err := weightNum.Float64()
	if err != nil {
		return errorInvalidParam
	}
	doSomething(id, name, weight)
	return nil
}

You can customize worker behavior by creating a goworker.WorkerSettings struct and passing it to goworker.SetSettings(settings). This allows you to configure the Redis URI, connection limits, concurrency, and specific queues.

package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func myFunc(queue string, args ...interface{}) error {
	fmt.Printf("From %s, %v\n", queue, args)
	return nil
}

func init() {
	settings := goworker.WorkerSettings{
		URI:            "redis://localhost:6379/",
		Connections:    100,
		Queues:         []string{"myqueue", "delimited", "queues"},
		UseNumber:      true,
		ExitOnComplete: false,
		Concurrency:    2,
		Namespace:      "resque:",
		Interval:       5.0,
	}
	goworker.SetSettings(settings)
	goworker.Register("MyClass", myFunc)
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}

You can programmatically enqueue jobs into a specific queue using goworker.Enqueue with a *goworker.Job object.

goworker.Enqueue(&goworker.Job{
    Queue: "myqueue",
    Payload: goworker.Payload{
        Class: "MyClass",
        Args: []interface{}{"hi", "there"},
    },
})

Use the WorkerSettings struct to define the behavior of the worker process. You can apply these settings using SetSettings(settings WorkerSettings) before calling Init() or Work().

Fields:

• QueuesString: A string representation of the queues to process.
• Queues: The parsed queue configuration.
• IntervalFloat: Float representation of the polling interval.
• Interval: The parsed polling interval.
• Concurrency: Number of concurrent workers to spawn.
• Connections: Number of Redis connections to maintain in the pool.
• URI: The Redis connection URI.
• Namespace: The Redis namespace.
• ExitOnComplete: If true, the process exits once queues are empty.
• IsStrict: Whether to use strict queue processing.
• UseNumber: Whether to use numeric identifiers.
• SkipTLSVerify: Whether to skip TLS verification.
• TLSCertPath: Path to the TLS certificate.

The Init() function initializes the goworker process, including the logger, flags, and the Redis connection pool.

While Work() calls this automatically, you can call Init() manually if you need to access goworker configuration or functions (like GetConn()) without starting the full worker loop. If you call Init() manually, you must call Close() to clean up resources.

if err := goworker.Init(); err != nil {
	fmt.Println("Error:", err)
}
defer goworker.Close()

The Close() function cleans up initialized resources, such as the Redis connection pool.

If you are using Work(), cleanup is handled automatically. If you are using Init() manually to access configuration or connections, you must call Close() to prevent resource leaks.