Adapters

A storage adapter is a type that implements some of VarMQ’s queue interfaces. redisq and sqliteq are adapters. A type of your own can also 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. If an interface is not implemented, its behaviour is off. This 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 shared storage does.

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 and IDistributedQueue. A priority column has the same three 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, ClosealwaysIQueueEnqueue, EnqueueOncenoIPriorityQueueEnqueue, EnqueueOncenoIAcknowledgeableDequeueWithAckId, AcknowledgenoISubscribableSubscribenoIPersistentQueueIQueue + IAcknowledgeablenoIPersistentPriorityQueueIPriorityQueue + IAcknowledgeablenoIDistributedQueueIPersistentQueue + ISubscribablenoIDistributedPriorityQueueIPersistentPriorityQueue + ISubscribablenoIBatchAcknowledgeableIAcknowledgeable + 1 methodDequeueBatchWithAckIdsnoCapacitySetterSetCapacityno

Accepted by

What VarMQ does with it

worker.WithQueue(q)

Start fromredisqsqliteqClear

What you pass and what you get

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 shared storage 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. 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 can hold the adapter behind an interface such as IQueue. Such a wrapper hides IBatchAcknowledgeable and CapacitySetter. The behaviour they add is then off.

The queue interfaces

IBaseQueue

The root of every queue. Every queue you give to VarMQ has these five methods:

On shared storage, Purge cannot close the jobs it drops, because they live in the storage. Instead, the handle stops waiting on the jobs it added, including running ones: their Err returns ErrDetached.

IQueue and IPriorityQueue

IQueue adds Enqueue(item any) bool and EnqueueOnce(item any, id string) bool to IBaseQueue. IPriorityQueue adds Enqueue(item any, priority int) bool and EnqueueOnce(item any, priority int, id string) bool.

The handle’s Add calls Enqueue once for each job, or EnqueueOnce when the job has an ID. true means the item is stored, or, on shared storage, that a job with the same ID already was. 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 give 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 three methods:

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 calls neither method for that job. Acknowledge or Reject returning false makes the worker report ErrAcknowledgeJob. The storage decides what happens to an item that is taken and not acknowledged. It can give it out again after a timeout, at the next start, or never. So the delivery guarantee is the adapter’s.

A worker takes one job per call. A job it took stays unacknowledged until it has run. If the worker held several, other consumers could not get them. 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 keeps none back, even if TunePool lowers the concurrency limit during the call. It also keeps none back if a pause or a stop arrives. The jobs are unacknowledged. If the worker kept them, a visibility timeout could give them to another consumer. A lower limit applies from the next call.
• n is how many it moved. The storage moves items into items[:n] and ackIDs[:n], each item under its own ID, in one atomic step. It moves no more than the smaller of the two slice lengths. 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 is empty. The worker then stops pulling until something signals work. Storage may also return a short call with remaining above 0. For example, it can do this for items it gives back from an earlier failed call. The worker then 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. Storage without Subscribe is signalled by the process’s own Add calls. Also, a worker that last saw the storage empty checks Len once a second. It does this while the worker is active and this is its only source. Storage with Subscribe is signalled by it. An adapter can implement both ISubscribable and this interface. Such an adapter has to send "enqueued" for items that 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.

EnqueueOnce

IQueue and IPriorityQueue both have EnqueueOnce. It stores the item unless an item with the same ID is already stored. A storage that does not check IDs can call its Enqueue from it.

Earlier releases had it in the optional IDedupQueue and IDedupPriorityQueue interfaces. Those are removed, so move the method onto your queue type as it is.

Add calls it, in place of Enqueue, for a job with an ID you gave, with WithJobId or Item.ID. A generated ID does not count. Job IDs says how long shared storage keeps an ID, and what Add returns for a repeat.

A producer made with NewQueue has no worker. The worker’s WithJobIdGenerator does not reach it. It uses an ID you pass with WithJobId, or the one DefaultJobIdGenerator sets. With neither, it makes a random ID when the storage reports outcomes or cancels jobs.

IPersistentQueue and IPersistentPriorityQueue

IPersistentQueue is IQueue plus IAcknowledgeable, and IPersistentPriorityQueue is IPriorityQueue plus IAcknowledgeable. They add no methods. NewQueue and NewPriorityQueue ask for them, and WithQueue and WithPriorityQueue detect them.

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.

WithQueue and WithPriorityQueue call Subscribe once, while binding, when shared storage (IAcknowledgeable) implements it. They 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. 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. NewQueue does not call Subscribe, because it has no worker.

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 the storage’s job 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.

Optional interfaces

Each interface below adds one feature. VarMQ checks for it on the storage value, and turns the feature off when it is missing. Most come from the contract module, which an adapter can import without VarMQ. The varmq package has an alias for each.

The types these methods use are in the contract module too.

• Outcomes: WatchOutcome reports a JobOutcome.
• Instances: a record is an InstanceRecord, with one InstanceWorker per worker.
• Scopes: a queue and a workflow store with the same ScopeID belong together. The adapter’s store constructor registers the store with contract.RegisterWorkflowStore, one store per scope in a process. VarMQ finds it with contract.LookupWorkflowStore. The store must be a pointer, and its Close must call contract.UnregisterWorkflowStore.
• Workflows: Submit takes a WorkflowPlan, with a TaskPlan per task and an UndoPlan per undo job. The store’s FinishTask takes the queue ID, the acknowledgement ID and a TaskEnd. Its Outcome is OutcomeCompleted, OutcomeFailed or OutcomeCancelled.
• Listing workflows: a store that implements WorkflowKeyLister lists WorkflowKey values. Then GET /workflows reads a full record only for the workflows on its page.

Details

The storage specification gives the exact rules for each method, such as how long a cancel mark lasts. The comments on each interface in the API reference say the same in short form.

The handle and the job interfaces

The rest of the interfaces are not for adapters to implement. They describe what VarMQ gives 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, Close, Pause, Resume and IsPaused. The handles a worker binds also have Worker(), and a producer handle from NewQueue does not. 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.

The handles of shared storage have them too. In the worker function, Job[T] is the argument for every queue, and on shared storage its ID() returns the id stored in the JSON. Jobs and worker types cover the handles.

What redisq and sqliteq implement

The table is for redisq v1.6.0 and sqliteq v0.4.0. Installation has the versions.

Next: REST API.