1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
create or replace function pgq.create_queue(i_queue_name text)
returns integer as $$
-- ----------------------------------------------------------------------
-- Function: pgq.create_queue(1)
--
-- Creates new queue with given name.
--
-- Returns:
-- 0 - queue already exists
-- 1 - queue created
-- ----------------------------------------------------------------------
declare
tblpfx text;
tblname text;
idxpfx text;
idxname text;
sql text;
id integer;
tick_seq text;
ev_seq text;
n_tables integer;
begin
if i_queue_name is null then
raise exception 'Invalid NULL value';
end if;
-- check if exists
perform 1 from pgq.queue where queue_name = i_queue_name;
if found then
return 0;
end if;
-- insert event
id := nextval('pgq.queue_queue_id_seq');
tblpfx := 'pgq.event_' || id;
idxpfx := 'event_' || id;
tick_seq := 'pgq.event_' || id || '_tick_seq';
ev_seq := 'pgq.event_' || id || '_id_seq';
insert into pgq.queue (queue_id, queue_name,
queue_data_pfx, queue_event_seq, queue_tick_seq)
values (id, i_queue_name, tblpfx, ev_seq, tick_seq);
select queue_ntables into n_tables from pgq.queue
where queue_id = id;
-- create seqs
execute 'CREATE SEQUENCE ' || tick_seq;
execute 'CREATE SEQUENCE ' || ev_seq;
-- create data tables
execute 'CREATE TABLE ' || tblpfx || ' () '
|| ' INHERITS (pgq.event_template)';
for i in 0 .. (n_tables - 1) loop
tblname := tblpfx || '_' || i;
idxname := idxpfx || '_' || i;
execute 'CREATE TABLE ' || tblname || ' () '
|| ' INHERITS (' || tblpfx || ')';
execute 'ALTER TABLE ' || tblname || ' ALTER COLUMN ev_id '
|| ' SET DEFAULT nextval(' || quote_literal(ev_seq) || ')';
execute 'create index ' || idxname || '_txid_idx on '
|| tblname || ' (ev_txid)';
end loop;
perform pgq.grant_perms(i_queue_name);
perform pgq.ticker(i_queue_name);
return 1;
end;
$$ language plpgsql security definer;
|
