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
|
create or replace function pgq.event_retry(
x_batch_id bigint,
x_event_id bigint,
x_retry_time timestamptz)
returns integer as $$
-- ----------------------------------------------------------------------
-- Function: pgq.event_retry(3)
--
-- Put the event into retry queue, to be processed again later.
--
-- Parameters:
-- x_batch_id - ID of active batch.
-- x_event_id - event id
-- x_retry_time - Time when the event should be put back into queue
--
-- Returns:
-- nothing
-- ----------------------------------------------------------------------
begin
insert into pgq.retry_queue (ev_retry_after,
ev_id, ev_time, ev_txid, ev_owner, ev_retry, ev_type, ev_data,
ev_extra1, ev_extra2, ev_extra3, ev_extra4)
select x_retry_time,
ev_id, ev_time, NULL, sub_id, coalesce(ev_retry, 0) + 1,
ev_type, ev_data, ev_extra1, ev_extra2, ev_extra3, ev_extra4
from pgq.get_batch_events(x_batch_id),
pgq.subscription
where sub_batch = x_batch_id
and ev_id = x_event_id;
if not found then
raise exception 'event not found';
end if;
return 1;
-- dont worry if the event is already in queue
exception
when unique_violation then
return 0;
end;
$$ language plpgsql security definer;
create or replace function pgq.event_retry(
x_batch_id bigint,
x_event_id bigint,
x_retry_seconds integer)
returns integer as $$
-- ----------------------------------------------------------------------
-- Function: pgq.event_retry(3)
--
-- Put the event into retry queue, to be processed later again.
--
-- Parameters:
-- x_batch_id - ID of active batch.
-- x_event_id - event id
-- x_retry_seconds - Time when the event should be put back into queue
--
-- Returns:
-- nothing
-- ----------------------------------------------------------------------
declare
new_retry timestamptz;
begin
new_retry := current_timestamp + ((x_retry_seconds || ' seconds')::interval);
return pgq.event_retry(x_batch_id, x_event_id, new_retry);
end;
$$ language plpgsql security definer;
|
