WIP. Using WIOService to handle Services

TODO

@@ -2,6 +2,7 @@
 [ServiceManager]
 - Rework the whole start/stop/try/cach/thread/interrupts things
 - Rework the io_service thread pool thing
+	- Use our own WIOService
 
 [Services]
 - [UI] generate argc/argv from a config file (crypto, port info, db path)

database-updater/DatabaseUpdater.cpp

@@ -2,6 +2,7 @@
 #include <boost/filesystem.hpp>
 #include <boost/foreach.hpp>
 #include <boost/thread.hpp>
+#include <boost/asio/placeholders.hpp>
 
 #include "database/MediaDirectory.hpp"
 #include "database/AudioTypes.hpp"
@@ -14,46 +15,114 @@ namespace DatabaseUpdater {
 using namespace Database;
 
 Updater::Updater(boost::filesystem::path dbPath, MetaData::Parser& parser)
- : _db(dbPath),
-   _metadataParser(parser)
+ : _running(false),
+_scheduleTimer(_ioService),
+_db(dbPath),
+_metadataParser(parser)
 {
-
+	_ioService.setThreadCount(1);
 }
 
 void
-Updater::process(void)
+Updater::start(void)
 {
-	removeMissingAudioFiles(_result.audioStats);
-	// TODO video files
+	_running = true;
 
+	// post some jobs in the io_service
+	processNextJob();
+
+	_ioService.start();
+}
+
+void
+Updater::stop(void)
+{
+	_running = false;
+
+	// TODO cancel all jobs (timer, ...)
+	_scheduleTimer.cancel();
+
+	_ioService.stop();
+}
+
+void
+Updater::processNextJob(void)
+{
 	Wt::Dbo::Transaction transaction(_db.getSession());
 
-	std::vector<MediaDirectory::pointer> mediaDirectories = MediaDirectory::getAll(_db.getSession());
+	MediaDirectorySettings::pointer settings = MediaDirectorySettings::get(_db.getSession());
 
-	BOOST_FOREACH( MediaDirectory::pointer directory, mediaDirectories)
+	if (settings->getManualScanRequested())
 	{
-		switch (directory->getType()) {
-			case MediaDirectory::Audio:
-				refreshAudioDirectory(directory->getPath(), _result.audioStats);
-				break;
-			case MediaDirectory::Video:
-				refreshVideoDirectory(directory->getPath());
-				break;
-		}
+		settings.modify()->setManualScanRequested(false);
+		// Schedule immediate scan
+		scheduleScan( boost::posix_time::seconds(0) );
 	}
+	else
+	{
+//		boost::posix_time::ptime now = boost::posix_time::second_clock::local_time();
 
-	std::cout << "Audio changes = " << _result.audioStats.nbChanges() << std::endl;
-	std::cout << "Video changes = " << _result.videoStats.nbChanges() << std::endl;
+		// TODO
+	}
+}
 
-	Database::MediaDirectorySettings::pointer settings = Database::MediaDirectorySettings::get(_db.getSession());
-	boost::posix_time::ptime now = boost::posix_time::second_clock::local_time();
+void
+Updater::scheduleScan( boost::posix_time::time_duration duration)
+{
+	std::cout << "Scheduling next scan in " << duration << std::endl;
+	_scheduleTimer.expires_from_now(duration);
+	_scheduleTimer.async_wait( boost::bind( &Updater::process, this, boost::asio::placeholders::error) );
+}
 
-	if (_result.audioStats.nbChanges() + _result.videoStats.nbChanges() > 0)
-		settings.modify()->setLastUpdate(now);
+void
+Updater::scheduleScan( boost::posix_time::ptime time)
+{
+	_scheduleTimer.expires_at(time);
+	_scheduleTimer.async_wait( boost::bind( &Updater::process, this, boost::asio::placeholders::error) );
+}
 
-	settings.modify()->setLastScan(now);
+void
+Updater::process(boost::system::error_code err)
+{
+	if (!err)
+	{
+		removeMissingAudioFiles(_result.audioStats);
+		// TODO video files
 
-	transaction.commit();
+		std::vector<boost::filesystem::path> pathes;
+
+		{
+			Wt::Dbo::Transaction transaction(_db.getSession());
+			std::vector<MediaDirectory::pointer> mediaDirectories = MediaDirectory::getAll(_db.getSession());
+			BOOST_FOREACH(MediaDirectory::pointer directory, mediaDirectories)
+				pathes.push_back(directory->getPath());
+		}
+
+		BOOST_FOREACH( boost::filesystem::path p, pathes)
+			refreshAudioDirectory(p, _result.audioStats);
+
+		std::cout << "Audio changes = " << _result.audioStats.nbChanges() << std::endl;
+		std::cout << "Video changes = " << _result.videoStats.nbChanges() << std::endl;
+
+		// Update database stats only if it has not been interrupted
+		if (_running)
+		{
+			boost::posix_time::ptime now = boost::posix_time::second_clock::local_time();
+			{
+				Wt::Dbo::Transaction transaction(_db.getSession());
+
+				Database::MediaDirectorySettings::pointer settings = Database::MediaDirectorySettings::get(_db.getSession());
+
+				if (_result.audioStats.nbChanges() + _result.videoStats.nbChanges() > 0)
+					settings.modify()->setLastUpdate(now);
+
+				settings.modify()->setLastScan(now);
+			}
+
+			processNextJob();
+		}
+
+	}
 }
 
 void
@@ -62,7 +131,7 @@ Updater::processAudioFile( const boost::filesystem::path& file, Stats& stats)
 	try {
 
 		// Check last update time
-	        boost::posix_time::ptime lastWriteTime (boost::posix_time::from_time_t( boost::filesystem::last_write_time( file ) ) );
+		boost::posix_time::ptime lastWriteTime (boost::posix_time::from_time_t( boost::filesystem::last_write_time( file ) ) );
 
 		Wt::Dbo::Transaction transaction(_db.getSession());
 
@@ -232,6 +301,12 @@ Updater::processAudioFile( const boost::filesystem::path& file, Stats& stats)
 void
 Updater::refreshAudioDirectory( const boost::filesystem::path& p, Stats& stats)
 {
+	if (!_running)
+	{
+		std::cerr << "Not running! Stopping scan" << std::endl;
+		return;
+	}
+
 	if (boost::filesystem::exists(p) && boost::filesystem::is_directory(p)) {
 
 		typedef std::vector<boost::filesystem::path> Paths;             // store paths,
@@ -247,6 +322,7 @@ Updater::refreshAudioDirectory( const boost::filesystem::path& p, Stats& stats)
 				if (boost::filesystem::is_directory(file)) {
 					refreshAudioDirectory( file, stats );
 				}
+
 				else if (boost::filesystem::is_regular(file)) {
 					processAudioFile( file, stats );
 				}

database-updater/DatabaseUpdater.hpp

@@ -1,6 +1,8 @@
 #ifndef DB_UPDATER_UPDATER_HPP
 #define DB_UPDATER_UPDATER_HPP
 
+#include <boost/asio/deadline_timer.hpp>
+#include <Wt/WIOService>
 
 #include "metadata/MetaData.hpp"
 #include "database/DatabaseHandler.hpp"
@@ -15,8 +17,9 @@ class Updater
 	public:
 		Updater(boost::filesystem::path db, MetaData::Parser& parser);
 
-		// Update database
-		void process();
+		void start();
+		void stop();
+
 
 	private:
 
@@ -36,6 +39,14 @@ class Updater
 			Stats	videoStats;
 		};
 
+		// Job handling
+		void processNextJob();
+		void scheduleScan(boost::posix_time::time_duration duration);
+		void scheduleScan(boost::posix_time::ptime time);
+
+		// Update database
+		void process(boost::system::error_code ec);
+
 		// Video
 		void refreshVideoDirectory( const boost::filesystem::path& directory );
 		void processVideoFile( const boost::filesystem::path& file);
@@ -47,6 +58,11 @@ class Updater
 
 		Database::Path::pointer getAddPath(const boost::filesystem::path& path);
 
+		bool			_running;
+		Wt::WIOService		_ioService;
+
+		boost::asio::deadline_timer _scheduleTimer;
+
 		Database::Handler	_db;
 
 		MetaData::Parser&	_metadataParser;

main/main.cpp

@@ -34,8 +34,8 @@ int main(int argc, char* argv[])
 
 		std::cout << "Starting services..." << std::endl;
 
-		serviceManager.startService( std::make_shared<DatabaseUpdateService>( serviceManager.getIoService(), dbPath) );
-		serviceManager.startService( std::make_shared<RemoteServerService>( serviceManager.getIoService(), remoteListenEndpoint, dbPath) );
+		serviceManager.startService( std::make_shared<DatabaseUpdateService>( dbPath) );
+		serviceManager.startService( std::make_shared<RemoteServerService>( remoteListenEndpoint, dbPath) );
 		serviceManager.startService( std::make_shared<UserInterfaceService>(argc, argv, dbPath) );
 
 		std::cout << "Running..." << std::endl;

remote/server/Server.cpp

@@ -9,14 +9,15 @@
 namespace Remote {
 namespace Server {
 
-Server::Server(boost::asio::io_service& ioService, const endpoint_type& bindEndpoint, boost::filesystem::path dbPath)
+Server::Server(const endpoint_type& bindEndpoint, boost::filesystem::path dbPath)
 :
-_ioService(ioService),
 _acceptor(_ioService, bindEndpoint, true /*SO_REUSEADDR*/),
 _connectionManager(),
 _context(boost::asio::ssl::context::tlsv1_server),
 _dbPath(dbPath)
 {
+	_ioService.setThreadCount(1);
+
 	_context.set_options( boost::asio::ssl::context::default_workarounds // TODO check this thing
 			| boost::asio::ssl::context::single_dh_use
 			| boost::asio::ssl::context::no_sslv2
@@ -29,12 +30,14 @@ _dbPath(dbPath)
 }
 
 void
-Server::run()
+Server::start()
 {
 	// While the server is running, there is always at least one
 	// asynchronous operation outstanding: the asynchronous accept call waiting
 	// for new incoming connections.
 	asyncAccept();
+
+	_ioService.start();
 }
 
 void
@@ -77,6 +80,8 @@ Server::stop()
 	// operations.
 	_acceptor.close();
 	_connectionManager.stopAll();
+
+	_ioService.stop();
 }
 
 } // namespace Server

remote/server/Server.hpp

@@ -1,6 +1,8 @@
 #ifndef REMOTE_SERVER_HPP
 #define REMOTE_SERVER_HPP
 
+#include <Wt/WIOService>
+
 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
 
@@ -24,10 +26,10 @@ class Server
 		typedef boost::asio::ip::tcp::endpoint 	endpoint_type;
 
 		// Serve up data from the given database
-		Server(boost::asio::io_service& ioService, const endpoint_type& bindEndpoint, boost::filesystem::path dbPath);
+		Server(const endpoint_type& bindEndpoint, boost::filesystem::path dbPath);
 
 		// Run the server's io_service loop.
-		void run();
+		void start();
 
 		void stop();
 
@@ -36,7 +38,7 @@ class Server
 		void asyncAccept();
 		void handleAccept(std::shared_ptr<Connection> newConnection, boost::system::error_code ec);
 
-		boost::asio::io_service& _ioService;
+		Wt::WIOService		_ioService;
 
 		/// Acceptor used to listen for incoming connections.
 		boost::asio::ip::tcp::acceptor _acceptor;

service/DatabaseUpdateService.cpp

@@ -2,7 +2,7 @@
 
 #include "DatabaseUpdateService.hpp"
 
-DatabaseUpdateService::DatabaseUpdateService(boost::asio::io_service& ioService, const boost::filesystem::path& p)
+DatabaseUpdateService::DatabaseUpdateService(const boost::filesystem::path& p)
 : _metadataParser(),
  _databaseUpdater( p, _metadataParser)
 {
@@ -11,18 +11,14 @@ DatabaseUpdateService::DatabaseUpdateService(boost::asio::io_service& ioService,
 void
 DatabaseUpdateService::start(void)
 {
-	// TODO
-	// Read database parameters and program a timer for the next scan
-//	_thread = boost::thread(boost::bind(&DatabaseUpdater::Updater::process, &_databaseUpdater));
+	_databaseUpdater.start();
 }
 
 void
 DatabaseUpdateService::stop(void)
 {
 	std::cout << "DatabaseUpdateService::stop, processing..." << std::endl;
-	// no effect if thread does not exist
-	_thread.interrupt();
-	_thread.join();
+	_databaseUpdater.stop();
 	std::cout << "DatabaseUpdateService::stop, process done" << std::endl;
 }
 
@@ -34,11 +30,3 @@ DatabaseUpdateService::restart(void)
 	start();
 }
 
-bool
-DatabaseUpdateService::isScanning(void) const
-{
-	// scanning is active only if a thread is running the updater
-	return _thread.get_id() != boost::thread::id();
-}
-
-

service/DatabaseUpdateService.hpp

@@ -15,20 +15,15 @@ class DatabaseUpdateService : public Service
 
 		typedef std::shared_ptr<DatabaseUpdateService>	pointer;
 
-		DatabaseUpdateService(boost::asio::io_service& ioService, const boost::filesystem::path& p);
+		DatabaseUpdateService(const boost::filesystem::path& p);
 
 		// Service interface
 		void start(void);
 		void stop(void);
 		void restart(void);
 
-		// Specific interface
-		bool	isScanning(void) const;	//return if the service is currently scanning the db
-
 	private:
 
-		boost::thread				_thread;
-
 		MetaData::AvFormat			_metadataParser;
 		DatabaseUpdater::Updater		_databaseUpdater;	// Todo use handler
 };

service/RemoteServerService.cpp

@@ -1,8 +1,8 @@
 
 #include "RemoteServerService.hpp"
 
-RemoteServerService::RemoteServerService(boost::asio::io_service& ioService, const Remote::Server::Server::endpoint_type& endpoint, boost::filesystem::path dbPath)
-: _server(ioService, endpoint, dbPath)
+RemoteServerService::RemoteServerService(const Remote::Server::Server::endpoint_type& endpoint, boost::filesystem::path dbPath)
+: _server(endpoint, dbPath)
 {
 }
 
@@ -10,7 +10,8 @@ void
 RemoteServerService::start(void)
 {
 	std::cout << "RemoteServerService::start, starting..." << std::endl;
-	_server.run();
+	_server.start();
+	std::cout << "RemoteServerService::start, started!" << std::endl;
 }
 
 
@@ -19,6 +20,7 @@ RemoteServerService::stop(void)
 {
 	std::cout << "RemoteServerService::stop, stopping..." << std::endl;
 	_server.stop();
+	std::cout << "RemoteServerService::stop, stopped!" << std::endl;
 }
 
 void

service/RemoteServerService.hpp

@@ -11,7 +11,7 @@ class RemoteServerService : public Service
 {
 	public:
 
-		RemoteServerService(boost::asio::io_service& ioService, const Remote::Server::Server::endpoint_type& endpoint, boost::filesystem::path dbPath);
+		RemoteServerService(const Remote::Server::Server::endpoint_type& endpoint, boost::filesystem::path dbPath);
 
 		void start(void);
 		void stop(void);

service/ServiceManager.cpp

@@ -27,7 +27,6 @@ ServiceManager::ServiceManager()
 
 ServiceManager::~ServiceManager()
 {
-	stopServices();
 }
 
 void
@@ -44,9 +43,11 @@ ServiceManager::run()
 	catch( std::exception& e )
 	{
 		std::cerr << "Caugh exception in service : " << e.what() << std::endl;
-		stopServices();
 	}
 
+	// Stopping services
+	stopServices();
+
 	std::cout << "ServiceManager::run complete!" << std::endl;
 }
 

service/ServiceManager.hpp

@@ -22,9 +22,6 @@ class ServiceManager
 		// Return in case of failure/stop by user
 		void run();
 
-		boost::asio::io_service&	getIoService() {return _ioService;}
-		const boost::asio::io_service&	getIoService() const {return _ioService;}
-
 		template <class T> typename T::pointer getService();
 
 		boost::mutex&	mutex() { return _mutex;}