Waiting

Add returns at once, and the worker runs the job later. A set of calls on the worker block until the worker as a whole reaches a point: drained, paused or stopped. They do not wait for one job. For that, use the job’s own handle (waiting on a job).

The scene below puts the wait calls of this page on one timeline. Pick a case, then press the button to let a blocked call go.

Wait and WaitUntilIdle

Wait and WaitUntilIdle both wait for the worker to finish its work. They differ in what they ask of the worker’s status:

In practice:

• On a worker that keeps running, both return when it has drained.
• If something may pause or stop the worker, Wait returns when the running jobs have finished, and jobs that have not started stay queued. WaitUntilIdle keeps waiting through a pause or a stop, until a later Resume, Restart or Start brings the worker back to Idle and it drains.
• A worker created with WithAutoRun(false) is Initiated until you call Start. Wait on it returns at once, with every job still queued. Start the worker first.

The program below pauses a worker while job 1 runs, with two more jobs queued behind it.

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(func(j varmq.Job[int]) {
		time.Sleep(100 * time.Millisecond)
		fmt.Println("job", j.Data(), "done")
	})
	queue := worker.BindQueue()

	for i := 1; i <= 3; i++ {
		queue.Add(i)
	}
	time.Sleep(10 * time.Millisecond) // job 1 is running
	worker.Pause()

	worker.Wait() // returns once job 1 is done
	fmt.Println("Wait returned - running:", worker.NumProcessing(), "- pending:", worker.NumPending())

	go func() {
		time.Sleep(500 * time.Millisecond)
		fmt.Println("resume")
		worker.Resume()
	}()

	worker.WaitUntilIdle() // waits through the pause
	fmt.Println("WaitUntilIdle returned - pending:", worker.NumPending())
}

Wait returns once job 1 is done, while NumProcessing is 0 and NumPending is still 2. WaitUntilIdle keeps waiting through the pause. It returns only after Resume has let jobs 2 and 3 run.

Both cover everything bound to the worker: each queue and each topic subscription. They return at a moment when the worker is empty, so a job added afterwards is not waited for.

Waiting with a timeout

Wait and the other wait calls take no context and no timeout. To give up after a time, run the call in a goroutine and select on it:

done := make(chan struct{})
go func() {
	worker.Wait()
	close(done)
}()

select {
case <-done:
	// Wait returned; on a paused or stopped worker, jobs can still be pending
case <-time.After(30 * time.Second):
	// gave up; the goroutine still waits until the work is done
}

The goroutine keeps waiting after the timeout. It ends when the worker drains, pauses or stops. Each call that times out leaves one goroutine waiting, so do not retry the timeout in a loop while the worker stays busy. When Wait returns, check NumPending and the status before you assume every job ran.

Waiting for a state

WaitUntilPaused and WaitUntilStopped wait for one status and look at nothing else. They return at once if the worker is already there. Otherwise they keep waiting until something puts it there, so a worker that you stop while you wait for Paused leaves the call waiting.

Pause and Stop return as soon as the change has started. While a job is running, the worker is Pausing or Stopping. Call WaitUntilPaused or WaitUntilStopped after them to wait until the change is over. The lifecycle page lists the states. Its first program shows a worker go from Pausing to Paused.

Combined calls

Four calls join a wait and a pause or a stop. The verb that comes first happens first.

Each returns an error: the one Pause or Stop returns. PauseAndWait and StopAndWait return it without waiting when the pause or stop failed. The lifecycle page lists the errors.

WaitAndStop drains, StopAndWait does not. On a worker that is running, WaitAndStop lets every pending job run, then starts the stop. A job added after Wait returns can still be left queued. So stop adding jobs before you call it. On a paused worker, Wait returns with jobs still pending, and they stay queued. It returns before the worker has reached Stopped, so call WaitUntilStopped after it if you need that. StopAndWait starts the stop at once and blocks until the worker is Stopped. Running jobs finish, and jobs that had not started stay queued: NumPending counts them, and they run after a Restart. See what happens to jobs.

The program below gives two workers three jobs each. The first gets StopAndWait while job 1 runs, and the second gets WaitAndStop, followed by WaitUntilStopped.

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	work := func(j varmq.Job[int]) {
		time.Sleep(100 * time.Millisecond)
		fmt.Println("job", j.Data(), "done")
	}

	stop := varmq.NewWorker(work)
	stopQueue := stop.BindQueue()
	for i := 1; i <= 3; i++ {
		stopQueue.Add(i)
	}
	time.Sleep(10 * time.Millisecond) // job 1 is running
	stop.StopAndWait()
	fmt.Println("StopAndWait:", stop.Status(), "- pending:", stop.NumPending())

	drain := varmq.NewWorker(work)
	drainQueue := drain.BindQueue()
	for i := 1; i <= 3; i++ {
		drainQueue.Add(i)
	}
	drain.WaitAndStop()
	drain.WaitUntilStopped()
	fmt.Println("WaitAndStop:", drain.Status(), "- pending:", drain.NumPending())
}

Under StopAndWait, job 1 finishes and jobs 2 and 3 stay pending. Under WaitAndStop, all three jobs run before the worker stops.

The pause pair works the same way. WaitAndPause lets the work finish before it pauses. PauseAndWait pauses at once and waits for the running jobs to end.

NumPending and NumProcessing

Two methods on Worker count the work in flight:

• NumPending is the number of jobs that have not started. It counts the jobs waiting in the bound queues and the jobs a goroutine has taken into a batch but not started. It also counts the jobs parked by a pause or stop. A queue’s own Len() counts only the first group. A job you closed while it waited still counts until the worker reaches it. While a goroutine works through a batch, the batch’s finished jobs still count until the whole batch ends. So during a run, it can be too high. It is exact once the worker is idle, paused or stopped.
• NumProcessing is the number of jobs running right now. It is at most the concurrency limit, except for a while after TunePool lowers it.

Each is read on its own, so the two together are not a snapshot of one moment. They suit a progress display or a backpressure check. To know that the work is finished, use Wait. For counters over the worker’s lifetime, see metrics.