Traite un événement MOVE_SET pour demander que l'origine de l'ensemble set_id soit déplacé du nœud old_origin vers le nœud new_origin.

declare
        p_set_id                        alias for $1;
        p_old_origin            alias for $2;
        p_new_origin            alias for $3;
        p_wait_seqno            alias for $4;
        v_local_node_id         int4;
        v_tab_row                       record;
        v_sub_row                       record;
        v_sub_node                      int4;
        v_sub_last                      int4;
        v_sub_next                      int4;
        v_last_sync                     int8;
begin
        -- ----
        -- Grab the central configuration lock
        -- ----
        lock table sl_config_lock;

        -- ----
        -- Get our local node ID
        -- ----
        v_local_node_id := getLocalNodeId('_schemadoc');

        -- ----
        -- If we are the old or new origin of the set, we need to
        -- remove the log trigger from all tables first.
        -- ----
        if v_local_node_id = p_old_origin or v_local_node_id = p_new_origin then
                for v_tab_row in select tab_id from sl_table
                                where tab_set = p_set_id
                                order by tab_id
                loop
                        perform alterTableRestore(v_tab_row.tab_id);
                end loop;
        end if;

        -- On the new origin, raise an event - ACCEPT_SET
        if v_local_node_id = p_new_origin then
                -- Create a SYNC event as well so that the ACCEPT_SET has
                -- the same snapshot as the last SYNC generated by the new
                -- origin. This snapshot will be used by other nodes to
                -- finalize the setsync status.
                perform createEvent('_schemadoc', 'SYNC', NULL);
                perform createEvent('_schemadoc', 'ACCEPT_SET', 
                        p_set_id::text, p_old_origin::text, 
                        p_new_origin::text, p_wait_seqno::text);
        end if;

        -- ----
        -- Next we have to reverse the subscription path
        -- ----
        v_sub_last = p_new_origin;
        select sub_provider into v_sub_node
                        from sl_subscribe
                        where sub_set = p_set_id
                        and sub_receiver = p_new_origin;
        if not found then
                raise exception 'Slony-I: subscription path broken in moveSet_int';
        end if;
        while v_sub_node <> p_old_origin loop
                -- ----
                -- Tracing node by node, the old receiver is now in
                -- v_sub_last and the old provider is in v_sub_node.
                -- ----

                -- ----
                -- Get the current provider of this node as next
                -- and change the provider to the previous one in
                -- the reverse chain.
                -- ----
                select sub_provider into v_sub_next
                                from sl_subscribe
                                where sub_set = p_set_id
                                        and sub_receiver = v_sub_node
                                for update;
                if not found then
                        raise exception 'Slony-I: subscription path broken in moveSet_int';
                end if;
                update sl_subscribe
                                set sub_provider = v_sub_last
                                where sub_set = p_set_id
                                        and sub_receiver = v_sub_node;

                v_sub_last = v_sub_node;
                v_sub_node = v_sub_next;
        end loop;

        -- ----
        -- This includes creating a subscription for the old origin
        -- ----
        insert into sl_subscribe
                        (sub_set, sub_provider, sub_receiver,
                        sub_forward, sub_active)
                        values (p_set_id, v_sub_last, p_old_origin, true, true);
        if v_local_node_id = p_old_origin then
                select coalesce(max(ev_seqno), 0) into v_last_sync 
                                from sl_event
                                where ev_origin = p_new_origin
                                        and ev_type = 'SYNC';
                if v_last_sync > 0 then
                        insert into sl_setsync
                                        (ssy_setid, ssy_origin, ssy_seqno,
                                        ssy_minxid, ssy_maxxid, ssy_xip, ssy_action_list)
                                        select p_set_id, p_new_origin, v_last_sync,
                                        ev_minxid, ev_maxxid, ev_xip, NULL
                                        from sl_event
                                        where ev_origin = p_new_origin
                                                and ev_seqno = v_last_sync;
                else
                        insert into sl_setsync
                                        (ssy_setid, ssy_origin, ssy_seqno,
                                        ssy_minxid, ssy_maxxid, ssy_xip, ssy_action_list)
                                        values (p_set_id, p_new_origin, '0',
                                        '0', '0', '', NULL);
                end if;
        end if;

        -- ----
        -- Now change the ownership of the set.
        -- ----
        update sl_set
                        set set_origin = p_new_origin
                        where set_id = p_set_id;

        -- ----
        -- On the new origin, delete the obsolete setsync information
        -- and the subscription.
        -- ----
        if v_local_node_id = p_new_origin then
                delete from sl_setsync
                                where ssy_setid = p_set_id;
        else
                if v_local_node_id <> p_old_origin then
                        --
                        -- On every other node, change the setsync so that it will
                        -- pick up from the new origins last known sync.
                        --
                        delete from sl_setsync
                                        where ssy_setid = p_set_id;
                        select coalesce(max(ev_seqno), 0) into v_last_sync
                                        from sl_event
                                        where ev_origin = p_new_origin
                                                and ev_type = 'SYNC';
                        if v_last_sync > 0 then
                                insert into sl_setsync
                                                (ssy_setid, ssy_origin, ssy_seqno,
                                                ssy_minxid, ssy_maxxid, ssy_xip, ssy_action_list)
                                                select p_set_id, p_new_origin, v_last_sync,
                                                ev_minxid, ev_maxxid, ev_xip, NULL
                                                from sl_event
                                                where ev_origin = p_new_origin
                                                        and ev_seqno = v_last_sync;
                        else
                                insert into sl_setsync
                                                (ssy_setid, ssy_origin, ssy_seqno,
                                                ssy_minxid, ssy_maxxid, ssy_xip, ssy_action_list)
                                                values (p_set_id, p_new_origin, '0',
                                                '0', '0', '', NULL);
                        end if;
                end if;
        end if;
        delete from sl_subscribe
                        where sub_set = p_set_id
                        and sub_receiver = p_new_origin;

        -- Regenerate sl_listen since we revised the subscriptions
        perform RebuildListenEntries();

        -- Run addPartialLogIndices() to try to add indices to unused sl_log_? table
        perform addPartialLogIndices();

        -- ----
        -- If we are the new or old origin, we have to
        -- put all the tables into altered state again.
        -- ----
        if v_local_node_id = p_old_origin or v_local_node_id = p_new_origin then
                for v_tab_row in select tab_id from sl_table
                                where tab_set = p_set_id
                                order by tab_id
                loop
                        perform alterTableForReplication(v_tab_row.tab_id);
                end loop;
        end if;

        return p_set_id;
end;