Jobs

A job is one unit of work: your data, an optional ID, and a status that changes as the worker deals with it. You never build one yourself. Add creates it, puts it in the queue and, on an in-memory queue, hands you a handle.

Two views of a job

The worker function and the code that called Add see different things.

The worker function gets a Job[T]. It has two methods: Data() returns what you passed to Add, and ID() returns the job’s ID. The interface has nothing else: no status and no Close.

The caller gets a handle. Add returns the handle and a bool:

job, ok := queue.Add("resize photo")

ok is false when the queue did not take the job: it is full or closed, or a job with the same ID is still in the queue (a job a worker has already taken into a batch, or parked, has released its ID). In that case job is nil, so check ok before you use it.

With a plain worker the handle is an EnqueuedJob. The other two worker types add one method that returns the outcome:

Worker	Add returns	Extra method
NewWorker	EnqueuedJob	none
NewErrWorker	EnqueuedErrJob	Err() error
NewResultWorker	EnqueuedResultJob[R]	Result() (R, error)

Every handle has ID(), Status(), IsClosed(), Wait() and Close(). Err and Result block until the job is closed, then return what the worker function sent. For a job that was closed before it ran, Err returns nil and Result returns the zero value and a nil error. See worker types for the three workers side by side.

Job IDs

A job has no ID unless you give it one. ID() returns an empty string by default.

• WithJobId sets the ID of one job: queue.Add(data, varmq.WithJobId("job-1")). An empty string is ignored.
• WithJobIdGenerator sets a function that makes IDs. Add calls it, on the caller’s goroutine, for every job; its value is used when the job has no ID of its own.

An in-memory queue refuses a second Add with an ID that a job still in the queue holds: it returns ok == false. A job a worker has already taken into a batch, or parked, has released its ID, so a duplicate Add is accepted even though that job’s status still reads Queued. IDs from a generator are not checked. Items added with AddAll get a g: prefix, so an item with ID a shows up in the worker function as g:a. Job IDs covers the details.

Job status

Status() on a handle returns one of five strings:

Status	Meaning
Created	The job exists but Add has not stored it yet. A handle you get back from Add is past this.
Queued	The job is waiting. It may be in the queue, in a batch a worker took but has not started, or parked by a pause or stop.
Processing	The worker function is running it.
Finished	The worker function returned.
Closed	The job is done with: it ran, or it was closed before it could.

Finished is short-lived. As soon as your function returns, the worker closes the job to release anyone waiting on it. By the time Wait returns, the status is Closed, and IsClosed() is true. So Closed does not tell you whether the job ran. Record that in your own worker function if it matters.

Move one job through its statuses below. Pick the worker type to see what the handle returns, and try Close() at each status.

One job · step through

status

Created

IsClosed

false

One job moves through the five statuses: Created, Queued, Processing, Finished and Closed. A worker type selector chooses what Add returns: a plain handle, one with Err, or one with Result. Buttons move the job: Add, the worker takes it, the function returns with success or an error, and Close. The scene shows what the caller's handle returns at each moment and what the worker function can see. Close fails with ErrJobProcessing on a running job and with ErrJobAlreadyClosed on a closed one.

1. Created
2. Queued
3. Processing
4. Finished
5. Closed

The caller's handle

The worker function gets Job[T]

j.Data()	"resize photo"
j.ID()	"job-1"
status, Close	not on the interface

worker := varmq.NewWorker(func(j varmq.Job[string]) { /* … */ })

Addjob.Close()Reset

worker

NewWorkerNewErrWorkerNewResultWorker

Finished lasts only an instant: the worker closes the job as soon as your function returns. A model of the library, not a benchmark.

The worker below runs one job at a time, and each job takes 500 ms. The program reads both statuses while job-1 runs.

jobs-status/main.go

package main

import (
  "fmt"
  "time"

  "github.com/goptics/varmq"
)

func main() {
  worker := varmq.NewWorker(func(j varmq.Job[string]) {
    fmt.Println("running", j.ID(), j.Data())
    time.Sleep(500 * time.Millisecond)
  })
  queue := worker.BindQueue()

  first, _ := queue.Add("resize photo", varmq.WithJobId("job-1"))
  second, _ := queue.Add("send email", varmq.WithJobId("job-2"))
  time.Sleep(100 * time.Millisecond) // job-1 has started by now

  fmt.Println(first.ID(), first.Status())
  fmt.Println(second.ID(), second.Status())

  first.Wait()
  second.Wait()
  fmt.Println(first.ID(), first.Status(), first.IsClosed())
  fmt.Println(second.ID(), second.Status(), second.IsClosed())
}

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

RunEditOpen in Go Playground

job-1 is Processing and job-2 is Queued. After Wait, both read Closed and IsClosed() is true, though both ran.

Closing a queued job

Close() cancels a job that has not started. It only marks the job closed; it does not take it out of the queue. When the worker reaches a closed job it skips it, and your function never sees it. Closing also releases Wait at once.

It succeeds while the job waits. It fails in two cases:

• A running job returns ErrJobProcessing. You cannot stop it from the outside.
• A closed job returns ErrJobAlreadyClosed. That includes a job that already ran, since the worker closes those too.

Until the worker reaches it, a job closed while it waited still counts in the queue’s Len() and in the worker’s NumPending(). It counts as submitted, and because it never ran, it never counts as completed.

Here job-1 is running and job-2 and job-3 wait behind it.

jobs-close/main.go

package main

import (
  "fmt"
  "time"

  "github.com/goptics/varmq"
)

func main() {
  worker := varmq.NewWorker(func(j varmq.Job[string]) {
    fmt.Println("running", j.ID(), j.Data())
    time.Sleep(500 * time.Millisecond)
  })
  queue := worker.BindQueue()

  first, _ := queue.Add("resize photo", varmq.WithJobId("job-1"))
  second, _ := queue.Add("send email", varmq.WithJobId("job-2"))
  third, _ := queue.Add("build report", varmq.WithJobId("job-3"))
  time.Sleep(100 * time.Millisecond) // job-1 is running, the others wait

  fmt.Println("close job-1:", first.Close())
  fmt.Println("close job-2:", second.Close())
  second.Wait() // returns at once: job-2 is closed
  fmt.Println("job-2 is", second.Status())
  fmt.Println("close job-2 again:", second.Close())

  third.Wait() // the worker skips job-2 and runs job-3
}

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

RunEditOpen in Go Playground

Closing job-1 fails with ErrJobProcessing. Closing job-2 works, its Wait returns at once while job-1 is still running, and a second Close fails with ErrJobAlreadyClosed. The worker then runs job-3 and never prints a line for job-2.

Waiting on a job

Wait() blocks until the job is closed, whether it ran or you closed it. If that has already happened, it returns at once. To wait for everything a worker has queued instead of one job, see waiting.