Ühel kenal päeval, E, 2007-03-26 kell 11:30, kirjutas Andrew Dunstan:
> This feature (ability to add a message payload to a NOTIFY) is on the
> TODO list and I had undertaken to implement it. However, pressure of
> other work has conspired to make that difficult, and Abhijit Menon-Sen
> recently very kindly offered to help out.
>
> Therer was some discussion of implementation late last year here:
> http://groups.google.com/group/pgsql.hackers/browse_frm/thread/e63a5ac43e2508ce/ce47016235bd5a62?tvc=1&q=notify+payload&hl=en#ce47016235bd5a62
>
> However, in various pieces of off-list discussion it appears that there
> is some opposition either to the design or to the feature itself. What
> is more, there will clearly be vigorous opposition to any implementation
> which does NOT remove the use of pg_listener (which I understand and I
> think largely agree with).
>
> So, before an investment of any more time is made by either Abhijit or
> myself, I would like to get confirmation that a) there is broad
> agreement on the desirability of the feature and b) that there is broad
> agreement on the general design (i.e. to use a circular buffer in shared
> memory, of configurable size, to hold the outstanding message queue).
> Please speak up or forever ....
I find the feature very useful, and have even done some preliminary
design work for shared memory implementation, where each listener is
required to copy data to its own privat memory ASAP and notifier waits
in case there is not enough room in shared memory buffer.
Alas, I have lost my design 3 a4 pages of design notes for organising
things in fixed-size buffer, but the basic operation data-wise sould be
as in attached sql file.
--
----------------
Hannu Krosing
Database Architect
Skype Technologies OÜ
Akadeemia tee 21 F, Tallinn, 12618, Estonia
Skype me: callto:hkrosing
Get Skype for free: http://www.skype.com
CREATE TABLE listen_name (
message_id serial PRIMARY KEY,
message_name text UNIQUE
);
CREATE TABLE listener (
listener_pid int,
message_id int REFERENCES listen_name,
CONSTRAINT pk_listener PRIMARY KEY (message_id, listener_pid)
);
CREATE TABLE message (
message_nr serial PRIMARY KEY,
message_id int REFERENCES listen_name,
notifier_pid int,
message_text text,
listener_count int
);
CREATE TABLE message_listener (
message_nr int REFERENCES message,
listener_pid int,
CONSTRAINT pk_message_listener PRIMARY KEY (listener_pid, message_nr)
);
CREATE LANGUAGE plpgsql;
CREATE FUNCTION add_listener (i_listener_pid int, i_message_name text) RETURNS VOID
LANGUAGE plpgsql AS
$$
/*
More error checking needed, what to do, when there is no room for adding a listener ?
*/
DECLARE
v_message_id int;
BEGIN
BEGIN
INSERT INTO listen_name(message_name) VALUES (i_message_name);
EXCEPTION WHEN unique_violation THEN
-- name already present, do nothing
END;
SELECT message_id INTO v_message_id FROM listen_name WHERE message_name = i_message_name;
INSERT INTO listener (listener_pid, message_id) VALUES (i_listener_pid, v_message_id);
END
$$
;
CREATE OR REPLACE FUNCTION add_message (i_notifier_pid int, i_message_name text, i_message_text text) RETURNS VOID
LANGUAGE plpgsql AS
$$
DECLARE
v_message_id int;
BEGIN
SELECT message_id INTO v_message_id FROM listen_name WHERE message_name = i_message_name;
IF NOT FOUND THEN
RETURN; -- nobody is listening, no need to post
END IF;
-- LOOP -- pseudocode for shared memory commented out
-- BEGIN
INSERT INTO message(message_id,notifier_pid,message_text,listener_count)
VALUES(v_message_id, i_notifier_pid, i_message_text, (SELECT count(*) FROM listener WHERE message_id = v_message_id));
INSERT INTO message_listener
SELECT currval('message_message_nr_seq'), listener_pid FROM listener WHERE message_id = v_message_id;
-- EXCEPTION WHEN not_enough_room_in_shared_mem THEN
-- -- wait for room to appear in shared mem, maybe try some cleanup
-- END;
-- END LOOP;
END
$$
;
CREATE OR REPLACE FUNCTION fetch_message (i_listener_pid int, i_message_name text, out o_notifier_pid int, out o_message_name text, out o_message_text text) RETURNS SETOF RECORD
LANGUAGE plpgsql AS
$$
DECLARE
v_message_id int;
v_message_nr int;
BEGIN
FOR v_message_nr, o_notifier_pid, o_message_name, o_message_text IN
SELECT ml.message_nr, notifier_pid,message_name,message_text
FROM message_listener ml, listen_name ln, message m
WHERE ml.listener_pid = i_listener_pid
AND m.message_id = ln.message_id
AND ml.message_nr = m.message_nr
ORDER BY message_nr
LOOP
UPDATE message SET listener_count = listener_count - 1 WHERE message_nr = v_message_nr;
DELETE FROM message WHERE message_nr = v_message_nr AND listener_count = 0; -- cleanup by the last listener
DELETE FROM message_listener WHERE (listener_pid, message_nr) = (i_listener_pid, v_message_nr);
RETURN NEXT;
END LOOP;
END
$$
;
CREATE FUNCTION drop_listener (i_listener_pid int) RETURNS VOID
LANGUAGE plpgsql AS
$$
UPDATE message SET listener_count = listener_count - 1 WHERE message_nr IN (SELECT message_nr FROM message_listener WHERE listener_pid = i_listener_pid);
DELETE FROM listener WHERE listener_pid = i_listener_pid; -- drop listeners of this pid
DELETE FROM listen_name WHERE message_id NOT IN (SELECT message_id FROM listener); -- drop message names with no registered listeners
DELETE FROM message_listener WHERE listener_pid = i_listener_pid; -- drop message listenings
DELETE FROM message WHERE listener_count = 0; -- clean up messages with no remaining listeners
$$
;
---------------------------(end of broadcast)---------------------------
TIP 7: You can help support the PostgreSQL project by donating at
http://www.postgresql.org/about/donate
