Priority queue

Queue that emits records ordered by an integer priority property instead of arrival order. Input records are stored and the call returns at once; a background worker emits them one by one on Output. Optionally persistable and duplicate-filtering, like the Reqular queue.

Pins

Pin	Kind	Types	Description
Input	input	any	Record to queue.
Clear	input	any	Any record; removes all queued records. The record content is not used.
Output	output	same as input	Queued records in priority order, one at a time, each with new metadata.
Already exists	continue	any	Records rejected as duplicates (same FilterProperty value within FilterLifetimeInMsec), with their original metadata.

Configuration

Field	Type	Default	Description
Name	string (max 128)	mypqueue	Required. Queue name; Queue writer components in the same flow write into the queue by this name.
PriorityProperty	dropdown (integer property of the input type)		Required. Property holding the priority. Lower value = emitted first. Options appear after the input is connected.
MaxItemsInQueue	integer (≥ 0)	0	Capacity, 0 = unlimited. Also bounds the duplicate filter.
FilterProperty	dropdown (property of the input type)		Optional. Property whose value is the duplicate key.
FilterLifetimeInMsec	integer (≥ 0)	0	How long a key is remembered after its record was queued, in milliseconds. 0 = no time limit.
Persistable	boolean	false	true keeps queued records and filter keys on disk and restores them when the flow starts again.
ItemLifetimeInSec	integer (≥ 0)	0	Records older than this (since they were queued) are dropped with a warning. 0 = never.

Behaviour

• At start the queue is created under Name for this flow and one worker starts. Without PriorityProperty the flow fails to start. The item type is the type of the Input connection (or of Output when only Queue writers feed the queue).
• Input: the priority is read from PriorityProperty and the record is added; the call returns (warm-up calls are ignored). Full queue: dropped with a warning. Duplicate: Already exists.
• Worker: always takes the record with the smallest priority value; equal priorities leave in arrival order. Expired records are skipped. The next record is taken only after the Output call returns, so a record queued later with a smaller value overtakes everything still waiting.
• Delivery is retried if the Output branch throws: again after 100 ms, doubling up to 30 s, until it succeeds, expires, the queue is cleared or the flow stops. Nothing is delivered while the flow is paused.
• Duplicates, Clear, persistence and queue naming work as in the Reqular queue: keys stay for FilterLifetimeInMsec also after the record left the queue; Clear empties records and keys (also on disk); a persistable queue restores its records (with their priorities) on start and discards them if the item type changed.

Notes

• Priority order is ascending: 0 before 1 before 10. Use negative values or invert your scale if "higher number = more important".
• Records leave the queue with new metadata.
• A record that always fails blocks the queue until it expires: set ItemLifetimeInSec or handle errors in the Output branch.
• A record whose priority property can't be read as an integer fails on Input.

Example

{ "Name": "jobs", "PriorityProperty": "Priority", "MaxItemsInQueue": 0, "FilterProperty": "JobId", "Persistable": true }

Http listener → Priority queue → (process job) → ...