Cascading replication option (#284)

* Option to enable cascade replication

* Requested changes

builds/install/misc/replication.conf

@@ -126,6 +126,13 @@ database
 
 	### REPLICA SIDE SETTINGS
 
+	# If enabled, changes applied to replica will be also subject of further
+	# replication (if any configured).
+	#
+	# cascade_replication = false
+
+	### Parameters specific for built-in replication
+
 	# Directory to search for the journal files to be replicated.
 	#
 	# journal_source_directory = 

src/jrd/replication/Applier.cpp

@@ -257,6 +257,9 @@ void Applier::process(thread_db* tdbb, ULONG length, const UCHAR* data)
 
 	tdbb->tdbb_flags |= TDBB_replicator;
 
+	const auto config = dbb->replConfig();
+	m_enableCascade = config != nullptr && config->cascadeReplication;
+
 	BlockReader reader(length, data);
 
 	const auto traNum = reader.getTransactionId();
@@ -834,6 +837,8 @@ void Applier::setSequence(thread_db* tdbb, const MetaName& genName, SINT64 value
 		attachment->att_generators.store(gen_id, genName);
 	}
 
+	AutoSetRestoreFlag<ULONG> noCascade(&tdbb->tdbb_flags, TDBB_repl_in_progress, !m_enableCascade);
+
 	if (DPM_gen_id(tdbb, gen_id, false, 0) < value)
 		DPM_gen_id(tdbb, gen_id, true, value);
 }
@@ -902,6 +907,7 @@ void Applier::executeSql(thread_db* tdbb,
 
   UserId* const owner = attachment->getUserId(ownerName);
 	AutoSetRestore<UserId*> autoOwner(&attachment->att_ss_user, owner);
+	AutoSetRestoreFlag<ULONG> noCascade(&tdbb->tdbb_flags, TDBB_repl_in_progress, !m_enableCascade);
 
 	DSQL_execute_immediate(tdbb, attachment, &transaction,
 						   0, sql.c_str(), dialect,
@@ -1178,6 +1184,7 @@ void Applier::doInsert(thread_db* tdbb, record_param* rpb, jrd_tra* transaction)
 
 	VIO_store(tdbb, rpb, transaction);
 	IDX_store(tdbb, rpb, transaction);
+	if (m_enableCascade)
 		REPL_store(tdbb, rpb, transaction);
 }
 
@@ -1275,6 +1282,7 @@ void Applier::doUpdate(thread_db* tdbb, record_param* orgRpb, record_param* newR
 
 	VIO_modify(tdbb, orgRpb, newRpb, transaction);
 	IDX_modify(tdbb, orgRpb, newRpb, transaction);
+	if (m_enableCascade)
 		REPL_modify(tdbb, orgRpb, newRpb, transaction);
 }
 
@@ -1287,6 +1295,7 @@ void Applier::doDelete(thread_db* tdbb, record_param* rpb, jrd_tra* transaction)
 	Savepoint::ChangeMarker marker(transaction->tra_save_point);
 
 	VIO_erase(tdbb, rpb, transaction);
+	if (m_enableCascade)
 		REPL_erase(tdbb, rpb, transaction);
 }
 

src/jrd/replication/Applier.h

@@ -154,6 +154,7 @@ namespace Jrd
 		Firebird::AutoPtr<RecordBitmap> m_bitmap;
 		Record* m_record;
 		JReplicator* m_interface;
+		bool m_enableCascade;
 
 		void startTransaction(thread_db* tdbb, TraNumber traNum);
 		void prepareTransaction(thread_db* tdbb, TraNumber traNum);

src/jrd/replication/Config.cpp

@@ -126,7 +126,8 @@ Config::Config()
 	  pluginName(getPool()),
 	  logErrors(true),
 	  reportErrors(false),
-	  disableOnError(true)
+	  disableOnError(true),
+	  cascadeReplication(false)
 {
 }
 
@@ -152,7 +153,8 @@ Config::Config(const Config& other)
 	  pluginName(getPool(), other.pluginName),
 	  logErrors(other.logErrors),
 	  reportErrors(other.reportErrors),
-	  disableOnError(other.disableOnError)
+	  disableOnError(other.disableOnError),
+	  cascadeReplication(other.cascadeReplication)
 {
 }
 
@@ -284,6 +286,10 @@ Config* Config::get(const PathName& lookupName)
 				{
 					parseBoolean(value, config->disableOnError);
 				}
+				else if (key == "cascade_replication")
+				{
+					parseBoolean(value, config->cascadeReplication);
+				}
 			}
 
 			if (exactMatch)

src/jrd/replication/Config.h

@@ -61,6 +61,7 @@ namespace Replication
 		bool logErrors;
 		bool reportErrors;
 		bool disableOnError;
+		bool cascadeReplication;
 	};
 };