pgmq

pig build pkg pgmq         # build RPM / DEB packages

pig repo add pgsql -u          # Add repo and update cache

CREATE EXTENSION pgmq;

CREATE EXTENSION pgmq;

SELECT pgmq.create('my_queue');

-- Send a single message (returns msg_id)
SELECT * FROM pgmq.send(
  queue_name => 'my_queue',
  msg        => '{"foo": "bar"}'
);

-- Send with delay (invisible for 5 seconds)
SELECT * FROM pgmq.send(
  queue_name => 'my_queue',
  msg        => '{"foo": "bar"}',
  delay      => 5
);

-- Send a batch of messages
SELECT pgmq.send_batch(
  queue_name => 'my_queue',
  msgs       => ARRAY['{"a":1}','{"b":2}','{"c":3}']::jsonb[]
);

SELECT * FROM pgmq.read(
  queue_name => 'my_queue',
  vt         => 30,    -- visibility timeout in seconds
  qty        => 2      -- number of messages to read
);

SELECT * FROM pgmq.pop('my_queue');

SELECT pgmq.delete('my_queue', 6);

SELECT pgmq.archive(queue_name => 'my_queue', msg_id => 2);

-- Archive multiple messages
SELECT pgmq.archive(queue_name => 'my_queue', msg_ids => ARRAY[3, 4, 5]);

SELECT * FROM pgmq.a_my_queue;

SELECT pgmq.drop_queue('my_queue');