Function helpers

A worker usually knows what its data means: a string is an email address, an int is a user ID, and the worker function knows what to do with it. When you only want to run pieces of code on a pool, the data can be a function instead. Func, ErrFunc and ResultFunc each return a ready-made worker function that calls the job’s function.

Helper	Use it with	The job’s data is	Add returns
Func()	NewWorker	func()	EnqueuedJob
ErrFunc()	NewErrWorker	func() error	EnqueuedErrJob
ResultFunc[R]()	NewResultWorker	func() (R, error)	EnqueuedResultJob[R]

You pass what the helper returns, so you call it:

worker := varmq.NewWorker(varmq.Func(), 4)
queue := worker.BindQueue()
queue.Add(func() { fmt.Println("hello") })

Go works out the data type from the helper, so there is nothing to spell out for Func and ErrFunc. ResultFunc returns a function for any result type, so you name it yourself: varmq.ResultFunc[string](). Worker types covers what each handle gives back.

Try it

The scene below shows what each helper wraps. Pick a helper and a closure, then add it. The lines that run are marked.

Function helpers · simulation

What each function helper wraps. Func, ErrFunc and ResultFunc each return a ready-made worker function. The worker function reads the job data, which is your closure, and calls it. If the closure is nil, Func panics, and the worker recovers the panic and reports it on Errs. ErrFunc returns an error, which Err returns. ResultFunc returns the zero value and an error, which Result returns. Pick a helper and a closure to see which lines of the helper run.

Func()ErrFunc()ResultFunc[string]()

a closure that worksa closure that failsnil

1. You add a closure. It becomes the job data.

3. What you get

Add returns

Errs()

2. The worker function the helper returns

queue := varmq.NewWorker(varmq.Func(), 1).BindQueue()

Add the closure

The program below runs all three helpers, each on a worker with concurrency 1.

function-helpers/main.go

package main

import (
  "errors"
  "fmt"

  "github.com/goptics/varmq"
)

func main() {
  // Func: the job's data is a func(), and the worker calls it.
  runner := varmq.NewWorker(varmq.Func(), 1)
  queue := runner.BindQueue()
  for i := 1; i <= 3; i++ {
    queue.Add(func() { fmt.Println("closure", i, "ran") })
  }
  runner.WaitUntilIdle()

  // ErrFunc: the data is a func() error. Err returns what it returned.
  checks := varmq.NewErrWorker(varmq.ErrFunc()).BindQueue()
  good, _ := checks.Add(func() error { return nil })
  bad, _ := checks.Add(func() error { return errors.New("disk full") })
  fmt.Println("ErrFunc, first:", good.Err())
  fmt.Println("ErrFunc, second:", bad.Err())

  // ResultFunc[R]: the data is a func() (R, error). Result returns both.
  fetches := varmq.NewResultWorker(varmq.ResultFunc[string]()).BindQueue()
  fetch, _ := fetches.Add(func() (string, error) { return "payload", nil })
  body, err := fetch.Result()
  fmt.Printf("ResultFunc: %q, %v\n", body, err)

  // A nil func is not a crash: the job fails with an error.
  empty, _ := checks.Add(nil)
  fmt.Println("nil func:", empty.Err())
}

Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.

RunEditOpen in Go Playground

The output, line by line:

1. closure 1 ran, closure 2 ran and closure 3 ran. One worker with concurrency 1 runs the queued functions in the order they were added. Each closure prints its own i, because from Go 1.22 on (when your go.mod says go 1.22 or later) every iteration of a for loop has its own variable.
2. ErrFunc, first: <nil> and ErrFunc, second: disk full. Err returned what each function returned.
3. ResultFunc: "payload", <nil>. Result returned both values of the function.
4. nil func: provided function is nil. The last job added nil as the function. The helper reports that as an error instead of calling it.

Details

A nil function. Each helper handles nil in its own way:

• Func panics. The worker recovers it, so the program carries on, and the panic goes to the worker’s Errs() channel, because a plain handle has no place for it.
• ErrFunc returns an error, which you get from Err().
• ResultFunc returns the zero value and an error, which you get from Result().

Panics. A panic inside your function is handled like a panic in any worker function. Worker types says where each kind reports it.

Closures run later. A function runs on a worker goroutine, not on the goroutine that called Add. Anything it captures has to be safe to share. Changing a captured variable after Add is a data race unless you synchronise the access.

No persistence. A persistent or distributed queue stores each job as JSON, and a function cannot be encoded as JSON. Keep function jobs for in-memory queues. For jobs that must survive a restart, use plain data and put the logic in the worker function.

Naming jobs. The job’s Data() is the function, so nothing about the job says what it does. WithJobId gives it a name. See job IDs.