replicapostgres.sql

java开源的企业总线.xmlBlaster

END;
$colToXmlCDATA$ LANGUAGE 'plpgsql';


-- ---------------------------------------------------------------------------- 
-- checkCreateDropAlter is used to check wether a table has been created,       
-- dropped or altered.                                                          
-- ---------------------------------------------------------------------------- 

CREATE OR REPLACE FUNCTION checkCreateDropAlter() 
   RETURNS text AS $checkCreateDropAlter$
   DECLARE colVar  RECORD;
           counter INT4;
BEGIN
    counter = 0;
    FOR colVar IN (SELECT relname FROM currentTables WHERE relname NOT IN (
               SELECT relname FROM pg_statio_user_tables WHERE relname IN 
	       (SELECT tableName FROM replTables)))
    LOOP
      INSERT INTO replitems (transaction, dbId, tableName, guid, action, 
                             schema, content, oldContent, version) values 
                             (CURRENT_TIMESTAMP, current_database(),
                             colVar.relname, colVar.oid, 'DROP', 
			     current_schema(), NULL, NULL, '0.0');
      counter = 1;
    END LOOP;					 
    FOR colVar IN (SELECT relname FROM pg_statio_user_tables WHERE relname 
	       IN (SELECT tableName FROM replTables) AND (relname NOT 
	       IN (SELECT relname FROM currentTables)))
      LOOP
      INSERT INTO replitems (transaction, dbId, tableName, guid, action, 
                             schema, content, oldContent, version) values 
                             (CURRENT_TIMESTAMP,current_database(),
                             colVar.relname, colVar.oid, 'CREATE', 
			     current_schema(), NULL, NULL, '0.0');
      counter = 1;
    END LOOP;
    IF counter > 0 THEN
       TRUNCATE currentTables;
       FOR colVar IN (SELECT relname FROM pg_statio_user_tables WHERE relname 
                  IN (SELECT tableName FROM replTables))
       LOOP 
         INSERT INTO currentTables (relname) VALUES (colVar.relname);
       END LOOP;
    --    the following would not work (probably because table is cached)         
    --    CREATE TABLE currentTables AS SELECT relname FROM pg_statio_user_tables;
    END IF;
    -- now CREATE and DROP have been handled ...                                  
    -- now we must handle ALTER TABLE (or column)                                 
    counter = 0;
    FOR colVar IN (SELECT relname FROM currentTables) LOOP
       IF isAltered(colVar.relname) = 't' THEN
         INSERT INTO replitems (transaction, dbId, tableName, guid, action, 
                                schema, content, oldContent, version) values 
                                (CURRENT_TIMESTAMP,current_database(),
                                colVar.relname, colVar.oid, 'ALTER', 
				current_schema(), NULL, NULL, '0.0');
	 counter = 1;
       END IF;
    END LOOP;
    IF counter > 0 THEN
      TRUNCATE replColsTable;
      FOR colVar IN (SELECT * FROM replColsView) LOOP
         INSERT INTO replColsTable VALUES (colVar.attname, colVar.owner, 
	                                   colVar.atttypid, colVar.attlen, 
					   colVar.attnotnull, colVar.attnum);
      END LOOP;
    END IF;
    RETURN 'OK';
END;
$checkCreateDropAlter$ LANGUAGE 'plpgsql';


-- ---------------------------------------------------------------------------- 
--                                                                              
-- BELOW THE BUSINESS SPECIFIC OPERATIONS AND FUNCTIONS ARE DEFINED             
--                                                                              
-- ---------------------------------------------------------------------------- 

-- ---------------------------------------------------------------------------- 
-- replTestGroup is the function which will be registered to the triggers.      
-- It must not take any parameter.                                              
-- This is the only method which is business data specific. It is depending on  
-- the table to be replicated. This should be generated by a tool.              
--                                                                              
-- For each table you should just write out in a sequence the complete content  
-- of the row to replicate. You could make more fancy stuff here, for example   
-- you could just send the minimal stuff, i.e. only the stuff which has changed 
-- (for the new stuff) and for the old one you could always send an empty one.  
-- ---------------------------------------------------------------------------- 

CREATE OR REPLACE FUNCTION replTestGroup() RETURNS trigger AS $replTestGroup$
DECLARE blobCont BYTEA; 
        oldCont TEXT; 
	newCont TEXT;
	comment TEXT;
	oid     TEXT;
	tmp     TEXT;
BEGIN
    oldCont = NULL;
    newCont = NULL;
    tmp = checkCreateDropAlter();

    IF (TG_OP != 'INSERT') THEN
-- this is needed since conversion from text to bytea must be done ...          
       blobCont = old.email; 
       oldCont = colToXml('uniqueId',old.uniqueId) || 
                 colToXml('name', old.name) || 
		 colToXmlCDATA('surname',old.surname) || 
		 colToXmlBASE64('email', blobCont) || 
		 colToXmlBASE64('photo', old.photo);
       oid = old.oid;
    END IF;

    IF (TG_OP != 'DELETE') THEN
       blobCont = new.email;
       newCont = colToXml('uniqueId',new.uniqueId) || 
                 colToXml('name', new.name) || 
		 colToXmlCDATA('surname',new.surname) || 
		 colToXmlBASE64('email', blobCont) || 
		 colToXmlBASE64('photo', new.photo);
       oid = new.oid;

    END IF;
    INSERT INTO replitems (transaction, dbId, tableName, guid, action, schema, 
                           content, oldContent, version) values 
                           (CURRENT_TIMESTAMP,current_database(),
			   TG_RELNAME, oid, TG_OP, current_schema(), newCont, 
			   oldCont, '0.0');
    tmp = inet_client_addr();
    comment = 'current user \'' || current_user || '\' session_user \'' ||
              session_user || '\' current schema \'' || current_schema() || '\'';
    RAISE NOTICE 'NEW REPL. ITEM %', comment;

    IF (TG_OP = 'DELETE') THEN RETURN OLD;
    END IF;
    RETURN NEW;
END;
$replTestGroup$ LANGUAGE 'plpgsql';

-- ---------------------------------------------------------------------------- 
-- replSystem is the function needed to detect CREATE TABLE and DROP TABLE      
-- operations. It is database specific but is needs to be created only once.    
-- It must not take any parameter.                                              
-- ---------------------------------------------------------------------------- 

CREATE OR REPLACE FUNCTION replSystem() RETURNS trigger AS $replSystem$
BEGIN
    IF (current_schema() != 'public') THEN
       RETURN NULL;
    END IF;

    IF (TG_OP != 'INSERT') THEN
       INSERT INTO replitems (transaction, dbId, tableName, guid, action, schema, 
                              content, oldContent, version) values 
                              (CURRENT_TIMESTAMP,current_database(),
     			      new.tablename, new.oid, 'CREATE', NULL, NULL, 
			      NULL, '0.0');
       RAISE NOTICE 'NEW REPL. ITEM: TABLE CREATED  %', new.tablename;
       RETURN NEW;
    END IF;

    IF (TG_OP != 'DELETE') THEN
       INSERT INTO replitems (transaction, dbId, tableName, guid, action, schema, 
                              content, oldContent, version) values 
                              (CURRENT_TIMESTAMP,current_database(),
     			      old.tablename, old.oid, 'DROP', NULL, NULL, 
			      NULL, '0.0');
       RAISE NOTICE 'NEW REPL. ITEM: TABLE DROPPED  %', old.tablename;
    
       RETURN OLD;
    END IF;


    IF (TG_OP = 'DELETE') THEN RETURN OLD;
    END IF;
    RETURN NEW;
END;
$replSystem$ LANGUAGE 'plpgsql';


-- ---------------------------------------------------------------------------- 
-- We need to create triggers. We make sure old triggers will be deleted before 
-- continuing.                                                                  
-- Note that the function invoked by a trigger can not take arguments (more on  
-- it on http://www.postgresql.org/docs/8.0/interactive/plpgsql-trigger.html).  
-- ---------------------------------------------------------------------------- 


-- ---------------------------------------------------------------------------- 
-- THE TRIGGER FOR THE replSystem FUNCTION                                      
-- ---------------------------------------------------------------------------- 


-- DROP TRIGGER triggerReplSystem ON pg_statio_user_tables CASCADE;
--CREATE TRIGGER triggerReplSystem
--AFTER DELETE OR INSERT
--ON pg_tables
--FOR EACH ROW
--EXECUTE PROCEDURE replSystem();			       



-- IMPORTANT: Rules do not have effect on table creation (no effect on pg_tables
-- nor on pg_statio_user_tables                                                 
CREATE OR REPLACE RULE tableChanges AS ON INSERT TO pg_attribute DO ALSO
      INSERT INTO replitems (transaction, dbId, tableName, guid, action, schema, 
                             content, oldContent, version) values 
                             (CURRENT_TIMESTAMP,current_database(),
    			      new.attname, NULL, 'CREATE', NULL, NULL, 
     		      NULL, '0.0');


-- ---------------------------------------------------------------------------- 
-- THE TRIGGER FOR THE replTest TABLE                                           
-- ---------------------------------------------------------------------------- 


DROP TRIGGER triggerReplTest ON replTest CASCADE;
CREATE TRIGGER triggerReplTest
AFTER UPDATE OR DELETE OR INSERT
ON repltest
FOR EACH ROW
EXECUTE PROCEDURE replTestGroup();

-- ---------------------------------------------------------------------------- 
-- THE TRIGGER FOR THE replTest2 TABLE                                          
-- ---------------------------------------------------------------------------- 

DROP TRIGGER triggerReplTest2 ON repltest2 CASCADE;
CREATE TRIGGER triggerReplTest2
AFTER UPDATE OR DELETE OR INSERT
ON repltest2
FOR EACH ROW
EXECUTE PROCEDURE replTestGroup();

-- ---------------------------------------------------------------------------- 
--  R E A D Y    T O   C L O S E   T R A N S A C T I O N                        
-- ---------------------------------------------------------------------------- 

COMMIT;


-- update currentTables after it has been detected that something changed
-- DROP TABLE currentTables;
-- CREATE TABLE currentTables AS SELECT relname FROM pg_statio_user_tables;

-- list of all tables which have been dropped since last update:
-- select relname from currentTables where relname NOT IN (select relname from pg_statio_user_tables);

-- list of all tables which have been created since last update (note the lowercases):
-- select relname from pg_statio_user_tables where relname NOT IN (select relname from currentTables) AND relname != 'currenttables';




-- FOR TESTING INVOKE:                                                          
-- update repltest SET email='[e-mail]' where name='laghi' AND uniqueId=1;  

-- ---------------------------------------------------------------------------- 
--                     E N D                                                    
-- ----------------------------------------------------------------------------