Adapters

A storage adapter is a type that implements some of varmq’s queue interfaces. redisq and sqliteq are adapters, and a type of your own can be one. varmq decides what a queue can do by the interfaces its value implements. It checks them with a type assertion when you bind the queue or add a job, so an interface that is not implemented switches its behaviour off and is not an error. This page lists them all, from IBaseQueue to the small job interfaces. They are all in the varmq package.

For using redisq or sqliteq, read Redis and SQLite instead. The overview covers what persistent and distributed queues do.

The map below shows how the interfaces fit together. Press a part to say that your type has it, and the boxes that combine parts switch on. Press a box to read what it adds, what the worker calls on it and when. The start-from buttons load redisq and sqliteq as the table at the end of the page describes them.

Interface map · simulation

A map of the varmq storage interfaces. The root is IBaseQueue, with five methods. A FIFO column has IQueue, IPersistentQueue, IDistributedQueue and IDedupQueue. A priority column has the same four with a priority. A third column has IAcknowledgeable, ISubscribable, IBatchAcknowledgeable and CapacitySetter. Each box is a button. Press a part to add it to your type. Press it again to remove it. Boxes that combine parts switch on by themselves. Below the map, one list shows which binding calls accept the type, and another list says what varmq does with it.

FIFO queue

Priority queue

Either

IBaseQueueLen, Dequeue, Values, Purge, ClosealwaysIQueueEnqueuenoIPriorityQueueEnqueuenoIAcknowledgeableDequeueWithAckId, AcknowledgenoISubscribableSubscribenoIPersistentQueueIQueue + IAcknowledgeablenoIPersistentPriorityQueueIPriorityQueue + IAcknowledgeablenoIDistributedQueueIPersistentQueue + ISubscribablenoIDistributedPriorityQueueIPersistentPriorityQueue + ISubscribablenoIDedupQueueEnqueueOncenoIDedupPriorityQueueEnqueueOncenoIBatchAcknowledgeableIAcknowledgeable + 1 methodDequeueBatchWithAckIdsnoCapacitySetterSetCapacityno

Accepted by

What varmq does with it

worker.WithPersistentQueue(pq)

Start fromredisq persistentredisq distributedsqliteqClear

What you pass and what you get

You call	Storage must implement	You get
WithQueue	IQueue	Queue[T]
WithPriorityQueue	IPriorityQueue	PriorityQueue[T]
WithPersistentQueue	IPersistentQueue	PersistentQueue[T]
WithPersistentPriorityQueue	IPersistentPriorityQueue	PersistentPriorityQueue[T]
WithDistributedQueue	IDistributedQueue	DistributedQueue[T]
WithDistributedPriorityQueue	IDistributedPriorityQueue	DistributedPriorityQueue[T]
NewDistributedQueue, no worker	IDistributedQueue	DistributedQueue[T]
NewDistributedPriorityQueue, no worker	IDistributedPriorityQueue	DistributedPriorityQueue[T]

Queues store values of type any. An in-memory queue, and one you bind with WithQueue or WithPriorityQueue, stores the job values varmq creates. Add on a persistent or distributed queue encodes the job as JSON and passes the []byte to Enqueue. What the storage gives back has to be that []byte, or a job value. Anything else is reported as ErrFailedToCastJob, and bytes that do not decode are ErrParseJob.

What the worker assumes of any queue:

• It is safe to call from several goroutines. Callers add jobs from their own goroutines, the worker’s event loop reads, and every job’s goroutine acknowledges when it ends.
• A read that finds nothing returns false for ok. The worker checks Len before it reads. If a read then finds nothing, for example because another consumer took the job meanwhile, the worker reports ErrFailedToDequeue on Errs() and to the handler.
• Optional interfaces are found on the value you pass. A wrapper that holds the adapter behind an interface such as IQueue hides IDedupQueue, IBatchAcknowledgeable and CapacitySetter, and the behaviour they add is off.

The queue interfaces

IBaseQueue

The root of every queue. Every queue you hand to varmq has these five methods:

Method	The worker calls it
Len() int	Whenever it decides whether there is work: before it pulls a job, for NumPending, for Wait, and for the MaxLen and MinLen strategies. On a persistent or distributed queue each call is a call to the storage.
Dequeue() (any, bool)	Once for each job it takes from a queue that does not acknowledge. It does not call it on a queue that is IAcknowledgeable, which it reads with DequeueWithAckId.
Values() []any	Only when you call Purge on the handle. varmq reads what is in the queue, clears it, then closes every value that implements io.Closer, which is how a purged job comes to read Closed.
Purge()	When you call Purge on the handle, as above.
Close() error	When you call Close on the handle.

On a handle from WithDistributedQueue or NewDistributedQueue, Len, Purge and Close are the adapter’s own methods, called directly. varmq adds nothing around them, and Purge closes no jobs.

IQueue and IPriorityQueue

IQueue adds Enqueue(item any) bool to IBaseQueue. IPriorityQueue adds Enqueue(item any, priority int) bool.

The handle’s Add calls Enqueue once for each job, unless the job has an ID and the queue implements IDedupQueue, which it calls instead. true means the item is stored. false makes Add return false.

The priority is the number the caller passed to Add. varmq does not order the queue: the worker takes whatever Dequeue returns next, so a priority queue has to hand out the lowest number first itself. For the built-in queues, and for redisq and sqliteq, a lower number comes first.

IAcknowledgeable

The interface that makes a queue persistent. It has two methods:

Method	The worker calls it
DequeueWithAckId() (any, bool, string)	Once for each job it takes, when a slot is free and Len is above 0. It returns the item, whether there was one, and an acknowledgement ID.
Acknowledge(ackID string) bool	Once for each job, with the ID it was given, when the job’s function has returned and the worker closes the job. It returns whether the storage accepted it.

The worker keeps the ID on the job and passes it back unchanged. An empty ID means there is nothing to acknowledge, and the worker does not call Acknowledge for that job. Acknowledge returning false makes the worker report ErrAcknowledgeJob. The storage decides what happens to an item that is taken and not acknowledged: it can hand it out again after a timeout, at the next start, or not at all. So the delivery guarantee is the adapter’s.

A worker takes one job per call, because a job it took stays unacknowledged until it has run, and holding several would keep them from other consumers. The exception is the batch interface below.

IBatchAcknowledgeable

An optional addition to IAcknowledgeable. IBatchAcknowledgeable has one more method, DequeueBatchWithAckIds(items []any, ackIDs []string) (n, remaining int, err error), for storage that can take many jobs in one call.

When the storage implements it and is the worker’s only bound source, the worker reads it differently:

• One call per wake-up. The worker counts its free slots and calls DequeueBatchWithAckIds with buffers of that size, at most 64. It does not call Len between jobs.
• Every job a call took starts. The worker starts each job as it decodes it. It holds none back, even if TunePool lowers the concurrency limit while the call is in flight, and even if a pause or a stop arrives. The jobs are unacknowledged, and holding them would let a visibility timeout hand them to another consumer. A lower limit applies from the next call.
• n is how many it moved. The storage moves up to the smaller of the two slice lengths into items[:n] and ackIDs[:n], each item under its own ID, in one atomic step. n is 0 with a nil error when the storage is empty.
• remaining is the length after the move. It may be a lower bound when the storage caps its count. It has to be above 0 whenever items remain, because the worker uses it to decide whether to call again.
• A short call normally means the storage ran dry. The worker then stops pulling until something signals work. Storage may also return a short call with remaining above 0, for example for items it hands back from an earlier failed call, and the worker pulls again.
• err is a failure. The worker reports it and pulls again on the next sign of work.
• An item that does not decode is reported and not acknowledged. The worker sends the error and leaves the item to the storage.

Signals matter because the worker does not probe the storage on its own after a short or failed call. A persistent queue is signalled by the process’s own Add calls. In addition, while the worker is active and this is its only source, a worker that last saw the storage empty checks Len once a second. A distributed queue is signalled by Subscribe. An adapter that implements ISubscribable and this interface has to send "enqueued" for items no Enqueue announced: items a visibility timeout put back, items another client added, items left after a failed call. Sending it about once a second while items wait does that.

With another source bound as well, the worker reads this storage one job at a time through DequeueWithAckId, and checks Len before each read.

Binding a storage that implements it also signals the worker, because the storage may already hold jobs and no Add will announce them. So the start-up trap does not arise with it.

IDedupQueue and IDedupPriorityQueue

IDedupQueue has EnqueueOnce(item any, id string) bool. IDedupPriorityQueue has EnqueueOnce(item any, priority int, id string) bool. Both are optional. EnqueueOnce stores the item unless an item with the same ID is already stored.

Add calls it, in place of Enqueue, for a job whose ID is not empty, when the storage implements it. Which IDs count depends on the queue:

• On a persistent or distributed queue, every ID counts, including one from a generator. Such storage keeps an ID for a window it sets. It returns true for a repeat too, so Add returns true while the job is stored once, and a persistent Add counts the repeat as submitted. That makes a retried Add safe after a lost reply. Use IDs that are unique across everything you add, such as a UUID, and not a business key or a counter that restarts.
• On an in-memory queue bound with WithQueue, only an ID you gave, with WithJobId or Item.ID, counts. The storage keeps an ID while its item is in the queue, until Dequeue or Purge takes it. A repeat returns false, and Add hands that back, because a job that was not stored would not finish. See job IDs.

A producer made with NewDistributedQueue has no worker. The worker’s WithJobIdGenerator does not reach it. It uses an ID you pass with WithJobId, or the one DefaultJobIdGenerator sets.

IPersistentQueue and IPersistentPriorityQueue

IPersistentQueue is IQueue plus IAcknowledgeable, and IPersistentPriorityQueue is IPriorityQueue plus IAcknowledgeable. They add no methods. They are what WithPersistentQueue and WithPersistentPriorityQueue ask for.

Binding one wraps it so that Add writes JSON and the worker reads with DequeueWithAckId, or with DequeueBatchWithAckIds when the storage implements it. The worker takes one job per free slot. Pause and Stop do not park a job it already took.

ISubscribable

ISubscribable has one method, Subscribe(func(action string)). It is what makes a queue distributed.

WithDistributedQueue and WithDistributedPriorityQueue call Subscribe once, while binding, and pass the worker’s handler. The adapter calls the handler with a string for each event. The worker acts on one: "enqueued" makes it wake to pull, and it also feeds the worker’s counters. Any other action is ignored, so an adapter may send others. The handler runs on whichever goroutine the adapter calls it from, and it does not block. The interface has no way to unsubscribe, so the handler stays until the adapter stops calling it. NewDistributedQueue does not call Subscribe, because it has no worker.

For a distributed queue, Submitted may differ from what you expect. These docs do not cover how a distributed queue’s counters count yet; Metrics covers in-memory queues only.

IDistributedQueue and IDistributedPriorityQueue

IDistributedQueue is IPersistentQueue plus ISubscribable, and IDistributedPriorityQueue is IPersistentPriorityQueue plus ISubscribable. They add no methods of their own.

CapacitySetter

CapacitySetter is optional and has one method, SetCapacity(cap int). varmq calls it when it creates a queue handle, with the number from WithQueueCapacity. It calls it even when you passed no capacity, with 0, so an implementation has to read a number below 1 as no limit. After that it is up to the storage to refuse Enqueue when it is full.

Without it, IsFull on the handle compares Len with the number, and Add is not refused. Queues covers capacity.

The handle and the job interfaces

The rest of the interfaces are not for adapters to implement. They describe what varmq hands you. They are listed here because they sit next to the queue interfaces in the API reference.

IExternalBaseQueue

IExternalBaseQueue is what every queue handle has: Len, IsFull, Purge and Close, and a deprecated NumPending that does what Len does. varmq implements it around your storage. Len, Purge and Close reach the IBaseQueue methods above, and IsFull reads Len.

Job interfaces

The job interfaces are small pieces that the job and group handles are built from. None of them is for storage to implement, because the jobs are varmq’s own values.

Interface	Methods	Where it appears
Identifiable	ID() string, empty when no ID was set	Job[T], the argument of the worker function, and every job handle
StatusProvider	IsClosed() bool and Status() string	every job handle. The worker calls IsClosed on each job it takes from a queue and skips a closed one.
Awaitable	Wait(), which blocks until the job is closed, or for a group until every job in it has ended	every job handle, and the group handles
Drainer	Drain(), which discards the results or errors a group’s channel still holds. A single job has no channel, so it does nothing there.	error and result job handles, and error and result group handles
PendingTracker	NumPending() int, the number of pending jobs in a group	the handles AddAll returns

They belong to the in-memory queues. A persistent or distributed Add returns only a bool, so none of these reaches you from it. In the worker function of a persistent queue, Job[T] is still the argument, and its ID() returns the id stored in the JSON. Jobs and worker types cover the handles.

What redisq and sqliteq implement

Interface	redisq	sqliteq
IBaseQueue, IQueue, IPriorityQueue, IAcknowledgeable	yes	yes
IPersistentQueue, IPersistentPriorityQueue	yes	yes
ISubscribable, IDistributedQueue, IDistributedPriorityQueue	yes, on the NewDistributed... queues	no
IBatchAcknowledgeable	yes	yes
IDedupQueue, IDedupPriorityQueue	yes	yes
CapacitySetter	no	no