Log Message:
-----------
Backpatch of slonik command SYNC.

Jan

Tags:
----
REL_1_1_STABLE

Modified Files:
--------------
    slony1-engine/src/ducttape:
        test_5_ddlscript (r1.2 -> r1.2.4.1)
        test_5_subscribe (r1.1 -> r1.1.4.1)
    slony1-engine/src/slonik:
        parser.y (r1.22.2.1 -> r1.22.2.2)
        scan.l (r1.23.2.1 -> r1.23.2.2)
        slonik.c (r1.42.2.4 -> r1.42.2.5)
        slonik.h (r1.23.2.1 -> r1.23.2.2)

-------------- next part --------------
Index: test_5_ddlscript
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/ducttape/test_5_ddlscript,v
retrieving revision 1.2
retrieving revision 1.2.4.1
diff -Lsrc/ducttape/test_5_ddlscript -Lsrc/ducttape/test_5_ddlscript -u -w -r1.2 -r1.2.4.1
--- src/ducttape/test_5_ddlscript
+++ src/ducttape/test_5_ddlscript
@@ -17,18 +17,7 @@
 # 
 ######################################################################
 
-cat >test_5_tmp.sql <<_EOF_
-	alter table accounts add column lastuser name;
-
-	create function accounts_lastuser () returns trigger as '
-	begin
-		NEW.lastuser = SESSION_USER;
-		return NEW;
-	end;
-	' language plpgsql;
-	create trigger accounts_lastuser before insert or update on accounts
-		for each row execute procedure accounts_lastuser();
-
+cat >test_5_tmp1.sql <<_EOF_
 	create table accounts_audit (
 		au_aid		integer,
 		au_seqno	serial,
@@ -40,6 +29,26 @@
 		foreign key (au_aid) references accounts
 	);
 
+	create table tellers_sum (
+		su_tid		integer,
+		su_sum		integer,
+
+		primary key (su_tid),
+		foreign key (su_tid) references tellers
+	);
+_EOF_
+cat >test_5_tmp2.sql <<_EOF_
+	alter table accounts add column lastuser name;
+
+	create function accounts_lastuser () returns trigger as '
+	begin
+		NEW.lastuser = SESSION_USER;
+		return NEW;
+	end;
+	' language plpgsql;
+	create trigger accounts_lastuser before insert or update on accounts
+		for each row execute procedure accounts_lastuser();
+
 	create function accounts_audit_trig () returns trigger as '
 	begin
 		--
@@ -58,13 +67,6 @@
 	create trigger accounts_audit_trig after update on accounts
 		for each row execute procedure accounts_audit_trig();
 
-	create table tellers_sum (
-		su_tid		integer,
-		su_sum		integer,
-
-		primary key (su_tid),
-		foreign key (su_tid) references tellers
-	);
 	insert into tellers_sum (su_tid, su_sum)
 		select tid, 0 from tellers;
 
@@ -81,7 +83,9 @@
 	node 2 admin conninfo = 'dbname=$DB2';
 
 	try {
-		execute script (set id = 1, filename = 'test_5_tmp.sql',
+		execute script (set id = 1, filename = 'test_5_tmp1.sql',
+			event node = 1);
+		execute script (set id = 1, filename = 'test_5_tmp2.sql',
 			event node = 1);
 
 		create set (id = 999, origin = 1, comment = 'temp set for new objects');
@@ -100,4 +104,4 @@
 	}
 _EOF_
 
-rm test_5_tmp.sql
+rm test_5_tmp[12].sql
Index: test_5_subscribe
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/ducttape/test_5_subscribe,v
retrieving revision 1.1
retrieving revision 1.1.4.1
diff -Lsrc/ducttape/test_5_subscribe -Lsrc/ducttape/test_5_subscribe -u -w -r1.1 -r1.1.4.1
--- src/ducttape/test_5_subscribe
+++ src/ducttape/test_5_subscribe
@@ -28,10 +28,15 @@
 	on error {
 		exit 1;
 	}
+	echo '***** Waiting for subscribe set to arrive on origin';
+	wait for event (origin = 2, confirmed = 1);
+	echo '***** Subscribe set received on origin - issue SYNC';
+	sync (id = 1);
+	echo '***** Waiting for SYNC to be done on subscriber';
+	wait for event (origin = 1, confirmed = 2);
+	echo '***** Subscription complete';
 _EOF_
 
-sleep 30
-
 echo "**** Merging set 999 into 1"
 slonik <<_EOF_
 	cluster name = T1;
Index: scan.l
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/slonik/scan.l,v
retrieving revision 1.23.2.1
retrieving revision 1.23.2.2
diff -Lsrc/slonik/scan.l -Lsrc/slonik/scan.l -u -w -r1.23.2.1 -r1.23.2.2
--- src/slonik/scan.l
+++ src/slonik/scan.l
@@ -123,6 +123,7 @@
 store			{ return K_STORE;			}
 subscribe		{ return K_SUBSCRIBE;			}
 success			{ return K_SUCCESS;			}
+sync			{ return K_SYNC;			}
 table			{ return K_TABLE;			}
 timeout			{ return K_TIMEOUT;			}
 trigger			{ return K_TRIGGER;			}
Index: slonik.h
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/slonik/slonik.h,v
retrieving revision 1.23.2.1
retrieving revision 1.23.2.2
diff -Lsrc/slonik/slonik.h -Lsrc/slonik/slonik.h -u -w -r1.23.2.1 -r1.23.2.2
--- src/slonik/slonik.h
+++ src/slonik/slonik.h
@@ -48,6 +48,7 @@
 typedef struct SlonikStmt_ddl_script_s SlonikStmt_ddl_script;
 typedef struct SlonikStmt_update_functions_s SlonikStmt_update_functions;
 typedef struct SlonikStmt_wait_event_s SlonikStmt_wait_event;
+typedef struct SlonikStmt_sync_s SlonikStmt_sync;
 
 typedef enum
 {
@@ -85,7 +86,8 @@
 	STMT_UNSUBSCRIBE_SET,
 	STMT_UPDATE_FUNCTIONS,
 	STMT_WAIT_EVENT,
-	STMT_ERROR
+	STMT_ERROR,
+	STMT_SYNC
 }	Slonik_stmttype;
 
 struct SlonikScript_s
@@ -423,6 +425,13 @@
 };
 
 
+struct SlonikStmt_sync_s
+{
+	SlonikStmt	hdr;
+	int			no_id;
+};
+
+
 
 
 extern SlonikScript *parser_script;
@@ -544,6 +553,7 @@
 extern int	slonik_ddl_script(SlonikStmt_ddl_script * stmt);
 extern int	slonik_update_functions(SlonikStmt_update_functions * stmt);
 extern int	slonik_wait_event(SlonikStmt_wait_event * stmt);
+extern int	slonik_sync(SlonikStmt_sync * stmt);
 
 extern int	slon_scanint64(char *str, int64 * result);
 
Index: parser.y
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/slonik/parser.y,v
retrieving revision 1.22.2.1
retrieving revision 1.22.2.2
diff -Lsrc/slonik/parser.y -Lsrc/slonik/parser.y -u -w -r1.22.2.1 -r1.22.2.2
--- src/slonik/parser.y
+++ src/slonik/parser.y
@@ -159,6 +159,7 @@
 %type <statement>	stmt_update_functions
 %type <statement>	stmt_repair_config
 %type <statement>	stmt_wait_event
+%type <statement>	stmt_sync
 %type <opt_list>	option_list
 %type <opt_list>	option_list_item
 %type <opt_list>	option_list_items
@@ -238,6 +239,7 @@
 %token	K_UPDATE
 %token	K_YES
 %token	K_WAIT
+%token	K_SYNC
 
 /*
  * Other scanner tokens
@@ -478,6 +480,8 @@
 					| stmt_error ';' 
 						{ yyerrok;
 						  $$ = $1; }
+					| stmt_sync
+						{ $$ = $1; }
 					;
 
 stmt_echo			: lno K_ECHO literal ';'
@@ -1443,6 +1447,32 @@
 					}
 					;
 
+stmt_sync			: lno K_SYNC option_list
+					{
+						SlonikStmt_sync *new;
+						statement_option opt[] = {
+							STMT_OPTION_INT( O_ID, -1 ),
+							STMT_OPTION_END
+						};
+
+						new = (SlonikStmt_sync *)
+								malloc(sizeof(SlonikStmt_sync));
+						memset(new, 0, sizeof(SlonikStmt_sync));
+						new->hdr.stmt_type		= STMT_SYNC;
+						new->hdr.stmt_filename	= current_file;
+						new->hdr.stmt_lno		= $1;
+
+						if (assign_options(opt, $3) == 0)
+						{
+							new->no_id			= opt[0].ival;
+						}
+						else
+							parser_errors++;
+
+						$$ = (SlonikStmt *)new;
+					}
+					;
+
 option_list			: ';'
 					{ $$ = NULL; }
 					| '(' option_list_items ')' ';'
Index: slonik.c
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/slonik/slonik.c,v
retrieving revision 1.42.2.4
retrieving revision 1.42.2.5
diff -Lsrc/slonik/slonik.c -Lsrc/slonik/slonik.c -u -w -r1.42.2.4 -r1.42.2.5
--- src/slonik/slonik.c
+++ src/slonik/slonik.c
@@ -1051,6 +1051,25 @@
 				}
 				break;
 
+			case STMT_SYNC:
+				{
+					SlonikStmt_sync *stmt =
+					(SlonikStmt_sync *) hdr;
+
+					if (stmt->no_id == -1)
+					{
+						printf("%s:%d: Error: "
+							   "node ID must be specified\n",
+							   hdr->stmt_filename, hdr->stmt_lno);
+						errors++;
+					}
+
+					if (script_check_adminfo(hdr, stmt->no_id) < 0)
+						errors++;
+
+				}
+				break;
+
 		}
 
 		hdr = hdr->next;
@@ -1476,6 +1495,16 @@
 				}
 				break;
 
+			case STMT_SYNC:
+				{
+					SlonikStmt_sync *stmt =
+					(SlonikStmt_sync *) hdr;
+
+					if (slonik_sync(stmt) < 0)
+						errors++;
+				}
+				break;
+
 		}
 
 		if (current_try_level == 0)
@@ -4056,6 +4085,36 @@
 }
 
 
+int
+slonik_sync(SlonikStmt_sync * stmt)
+{
+	SlonikAdmInfo *adminfo1;
+	SlonDString query;
+
+	adminfo1 = get_active_adminfo((SlonikStmt *) stmt, stmt->no_id);
+	if (adminfo1 == NULL)
+		return -1;
+
+	if (db_begin_xact((SlonikStmt *) stmt, adminfo1) < 0)
+		return -1;
+
+	dstring_init(&query);
+
+	slon_mkquery(&query,
+				 "select \"_%s\".createEvent('_%s', 'SYNC'); ",
+				 stmt->hdr.script->clustername,
+				 stmt->hdr.script->clustername);
+	if (db_exec_command((SlonikStmt *) stmt, adminfo1, &query) < 0)
+	{
+		dstring_free(&query);
+		return -1;
+	}
+
+	dstring_free(&query);
+	return 0;
+}
+
+
 /*
  * scanint8 --- try to parse a string into an int8.
  *
@@ -4122,6 +4181,52 @@
 	return true;
 }
 
+
+/*
+ * make a copy of the array of lines, with token replaced by replacement
+ * the first time it occurs on each line.
+ *
+ * This does most of what sed was used for in the shell script, but
+ * doesn't need any regexp stuff.
+ */
+static void
+replace_token(char *resout, char *lines, const char *token, const char *replacement)
+{
+	int			numlines = 1;
+	int			i,
+				o;
+	char		result_set[4096];
+	int			toklen,
+				replen;
+
+	for (i = 0; lines[i]; i++)
+		numlines++;
+
+	toklen = strlen(token);
+	replen = strlen(replacement);
+
+	for (i = o = 0; i < numlines; i++, o++)
+	{
+		/* just copy pointer if NULL or no change needed */
+		if (!lines[i] || (strncmp((const char *)lines + i, token, toklen)))
+		{
+			if (lines[i] == 0x0d)		/* ||(lines[i] == 0x0a)) */
+				break;
+
+			result_set[o] = lines[i];
+			continue;
+		}
+		/* if we get here a change is needed - set up new line */
+		strncpy((char *)result_set + o, replacement, replen);
+		o += replen - 1;
+		i += toklen - 1;
+	}
+
+	result_set[o] = '\0';
+	memcpy(resout, result_set, o);
+}
+
+
 /*
  * Local Variables:
  *	tab-width: 4