Centralise and wrap maintenance Tx management

9 files changed, 38 insertions, 15 deletions

diff --git a/p2pvr/Jamfile.jam b/p2pvr/Jamfile.jam
index d709f2d..b1190f6 100644
--- a/p2pvr/Jamfile.jam
+++ b/p2pvr/Jamfile.jam
@@ -13,6 +13,10 @@ alias p2sql : glibmm : : :
 	<include>/usr/include/project2/sql
 	<linkflags>"-lp2sql"
 ;
+alias p2lib : glibmm : : :
+	<include>/usr/include/project2/lib
+	<linkflags>"-lp2lib"
+;
 alias p2daemonlib : glibmm : : :
 	<cflags>"-I /usr/include/project2/daemon/lib"
 	<linkflags>"-lp2daemonlib"
diff --git a/p2pvr/lib/Jamfile.jam b/p2pvr/lib/Jamfile.jam
index 9814478..2947d05 100644
--- a/p2pvr/lib/Jamfile.jam
+++ b/p2pvr/lib/Jamfile.jam
@@ -10,6 +10,7 @@ lib p2pvrlib :
 	<library>../ice//p2pvrice
 	<library>..//p2common
 	<library>..//p2sql
+	<library>..//p2lib
 	<implicit-dependency>../ice//p2pvrice
 	: :
 	<library>boost_filesystem
diff --git a/p2pvr/lib/maintenance.cpp b/p2pvr/lib/maintenance.cpp
index 540b323..d644d22 100644
--- a/p2pvr/lib/maintenance.cpp
+++ b/p2pvr/lib/maintenance.cpp
@@ -31,3 +31,18 @@ Maintenance::UpdateAll(short type, const Ice::Current & ice)
 	UpdateEvents(type, ice);
 }
 
+void
+Maintenance::onAllDatasources(const DataSourceCall & call) const
+{
+	BOOST_FOREACH(const CommonObjects::DataSources::value_type & ds, CommonObjects::datasources) {
+		call(ds.second);
+	}
+}
+
+Maintenance::TxHelper::TxHelper(const Maintenance * m) :
+	so(NULL,
+		boost::bind(&Maintenance::onAllDatasources, m, DataSourceCall(boost::bind(&DataSource::commit, _1))),
+		boost::bind(&Maintenance::onAllDatasources, m, DataSourceCall(boost::bind(&DataSource::rollback, _1))))
+{
+}
+
diff --git a/p2pvr/lib/maintenance.h b/p2pvr/lib/maintenance.h
index b1965d1..d1b36e1 100644
--- a/p2pvr/lib/maintenance.h
+++ b/p2pvr/lib/maintenance.h
@@ -3,11 +3,19 @@
 
 #include <p2pvr.h>
 #include <commonObjects.h>
+#include <scopeObject.h>
 
 class SqlMergeTask;
 
 class Maintenance : public P2PVR::Maintenance, public virtual CommonObjects {
 	public:
+		class TxHelper {
+			public:
+				TxHelper(const Maintenance *);
+			private:
+				ScopeObject so;
+		};
+
 		void UpdateAll(const Ice::Current &);
 		void UpdateAll(short type, const Ice::Current &);
 		void UpdateNetwork(short type, const Ice::Current &);
@@ -17,6 +25,11 @@ class Maintenance : public P2PVR::Maintenance, public virtual CommonObjects {
 		void UpdateEvents(short type, const Ice::Current &);
 
 		static void SqlMergeColumnsInserter(SqlMergeTask * merge, const std::string & name, bool key);
+
+	private:
+		friend class TxHelper;
+		typedef boost::function<void(DataSourcePtr)> DataSourceCall;
+		void onAllDatasources(const DataSourceCall &) const;
 };
 
 #endif
diff --git a/p2pvr/lib/maintenance/events.cpp b/p2pvr/lib/maintenance/events.cpp
index 7edac72..67b8a74 100644
--- a/p2pvr/lib/maintenance/events.cpp
+++ b/p2pvr/lib/maintenance/events.cpp
@@ -75,14 +75,12 @@ class SiEventsMerger : public IHaveSubTasks {
 void
 Maintenance::UpdateEvents(short type, const Ice::Current & ice)
 {
+	TxHelper tx(this);
 	SqlMergeTask mergeEvents("postgres", "events");
 	CreateColumns<DVBSI::EventPtr>(boost::bind(SqlMergeColumnsInserter, &mergeEvents, _1, _2));
 	mergeEvents.sources.insert(new SiEventsMerger(type, ice));
 	mergeEvents.loadComplete(this);
 	mergeEvents.execute(NULL);
-	BOOST_FOREACH(const CommonObjects::DataSources::value_type & ds, CommonObjects::datasources) {
-		ds.second->commit();
-	}
 	Logger()->messagebf(LOG_INFO, "%s: Updated events", __PRETTY_FUNCTION__);
 }
 
diff --git a/p2pvr/lib/maintenance/network.cpp b/p2pvr/lib/maintenance/network.cpp
index f49e3f2..9947d72 100644
--- a/p2pvr/lib/maintenance/network.cpp
+++ b/p2pvr/lib/maintenance/network.cpp
@@ -74,6 +74,7 @@ Maintenance::UpdateNetwork(short type, const Ice::Current & ice)
 		throw std::runtime_error("bad proxy(s)");
 	}
 	const auto transports = si->GetAllDeliveries(type);
+	TxHelper tx(this);
 	// Attempt to just download fresh data
 	BOOST_FOREACH(const auto & transport, transports) {
 		P2PVR::TunerPrx tuner;
@@ -90,9 +91,6 @@ Maintenance::UpdateNetwork(short type, const Ice::Current & ice)
 		try {
 			tuner->SendNetworkInformation(parser);
 			devs->ReleaseTuner(tuner);
-			BOOST_FOREACH(const CommonObjects::DataSources::value_type & ds, CommonObjects::datasources) {
-				ds.second->commit();
-			}
 			return;
 		}
 		catch (const std::exception & ex) {
diff --git a/p2pvr/lib/maintenance/programAssociations.cpp b/p2pvr/lib/maintenance/programAssociations.cpp
index 47e48b3..e5913a3 100644
--- a/p2pvr/lib/maintenance/programAssociations.cpp
+++ b/p2pvr/lib/maintenance/programAssociations.cpp
@@ -65,6 +65,7 @@ Maintenance::UpdateProgramAssociations(short type, const Ice::Current & ice)
 		}
 	}
 
+	TxHelper tx(this);
 	SqlMergeTask mergeServices("postgres", "services");
 	CreatePATColumns(boost::bind(SqlMergeColumnsInserter, &mergeServices, _1, _2));
 	// Don't change the list of services available from the network
@@ -74,8 +75,5 @@ Maintenance::UpdateProgramAssociations(short type, const Ice::Current & ice)
 	mergeServices.sources.insert(new MapIterator<ProgramAssociationMap>(CreatePATColumns, &siparser->map));
 	mergeServices.loadComplete(this);
 	mergeServices.execute(NULL);
-	BOOST_FOREACH(const CommonObjects::DataSources::value_type & ds, CommonObjects::datasources) {
-		ds.second->commit();
-	}
 }
 
diff --git a/p2pvr/lib/maintenance/programMap.cpp b/p2pvr/lib/maintenance/programMap.cpp
index bd4c2ee..7967872 100644
--- a/p2pvr/lib/maintenance/programMap.cpp
+++ b/p2pvr/lib/maintenance/programMap.cpp
@@ -122,13 +122,11 @@ class SiProgramMapMerger : public IHaveSubTasks {
 void
 Maintenance::UpdateProgramMaps(short type, const Ice::Current & ice)
 {
+	TxHelper tx(this);
 	SqlMergeTask mergeServiceStreams("postgres", "servicestreams");
 	CreateColumns<DVBSI::StreamPtr>(boost::bind(SqlMergeColumnsInserter, &mergeServiceStreams, _1, _2));
 	mergeServiceStreams.sources.insert(new SiProgramMapMerger(type, this, ice));
 	mergeServiceStreams.loadComplete(this);
 	mergeServiceStreams.execute(NULL);
-	BOOST_FOREACH(const CommonObjects::DataSources::value_type & ds, CommonObjects::datasources) {
-		ds.second->commit();
-	}
 }
 
diff --git a/p2pvr/lib/maintenance/services.cpp b/p2pvr/lib/maintenance/services.cpp
index 08b8c9d..e7e646d 100644
--- a/p2pvr/lib/maintenance/services.cpp
+++ b/p2pvr/lib/maintenance/services.cpp
@@ -56,14 +56,12 @@ Maintenance::UpdateServices(short type, const Ice::Current & ice)
 		throw std::runtime_error("no delivery methods");
 	}
 
+	TxHelper tx(this);
 	Logger()->messagebf(LOG_DEBUG, "%s: Getting a tuner", __PRETTY_FUNCTION__);
 	auto tuner = devs->GetTunerAny(type, deliveries.front(), time(NULL) + 90);
 	Logger()->messagebf(LOG_DEBUG, "%s: Fetching service list", __PRETTY_FUNCTION__);
 	tuner->SendServiceDescriptions(parser);
 	Logger()->messagebf(LOG_INFO, "%s: Updated service list", __PRETTY_FUNCTION__);
 	devs->ReleaseTuner(tuner);
-	BOOST_FOREACH(const CommonObjects::DataSources::value_type & ds, CommonObjects::datasources) {
-		ds.second->commit();
-	}
 }