From 9fdc9f41dd0157033373e18abb50dfd5d0bb12c1 Mon Sep 17 00:00:00 2001 From: emeric Date: Tue, 5 Aug 2014 13:11:31 +0200 Subject: [PATCH] WIP. Using WIOService to handle Services --- TODO | 1 + database-updater/DatabaseUpdater.cpp | 126 +++++++++++++++++++++------ database-updater/DatabaseUpdater.hpp | 20 ++++- main/main.cpp | 4 +- remote/server/Server.cpp | 11 ++- remote/server/Server.hpp | 8 +- service/DatabaseUpdateService.cpp | 18 +--- service/DatabaseUpdateService.hpp | 7 +- service/RemoteServerService.cpp | 8 +- service/RemoteServerService.hpp | 2 +- service/ServiceManager.cpp | 5 +- service/ServiceManager.hpp | 3 - 12 files changed, 148 insertions(+), 65 deletions(-) diff --git a/TODO b/TODO index 3b93964a..b0801c06 100644 --- a/TODO +++ b/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) diff --git a/database-updater/DatabaseUpdater.cpp b/database-updater/DatabaseUpdater.cpp index 3ccd4dc8..a1ae3d75 100644 --- a/database-updater/DatabaseUpdater.cpp +++ b/database-updater/DatabaseUpdater.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #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 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 pathes; + + { + Wt::Dbo::Transaction transaction(_db.getSession()); + std::vector 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 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 ); } diff --git a/database-updater/DatabaseUpdater.hpp b/database-updater/DatabaseUpdater.hpp index 61487dda..c27b2896 100644 --- a/database-updater/DatabaseUpdater.hpp +++ b/database-updater/DatabaseUpdater.hpp @@ -1,6 +1,8 @@ #ifndef DB_UPDATER_UPDATER_HPP #define DB_UPDATER_UPDATER_HPP +#include +#include #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; diff --git a/main/main.cpp b/main/main.cpp index 391842c6..9edaf2bb 100644 --- a/main/main.cpp +++ b/main/main.cpp @@ -34,8 +34,8 @@ int main(int argc, char* argv[]) std::cout << "Starting services..." << std::endl; - serviceManager.startService( std::make_shared( serviceManager.getIoService(), dbPath) ); - serviceManager.startService( std::make_shared( serviceManager.getIoService(), remoteListenEndpoint, dbPath) ); + serviceManager.startService( std::make_shared( dbPath) ); + serviceManager.startService( std::make_shared( remoteListenEndpoint, dbPath) ); serviceManager.startService( std::make_shared(argc, argv, dbPath) ); std::cout << "Running..." << std::endl; diff --git a/remote/server/Server.cpp b/remote/server/Server.cpp index 88d21459..7726629a 100644 --- a/remote/server/Server.cpp +++ b/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 diff --git a/remote/server/Server.hpp b/remote/server/Server.hpp index 2a70fb8f..07634762 100644 --- a/remote/server/Server.hpp +++ b/remote/server/Server.hpp @@ -1,6 +1,8 @@ #ifndef REMOTE_SERVER_HPP #define REMOTE_SERVER_HPP +#include + #include #include @@ -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 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; diff --git a/service/DatabaseUpdateService.cpp b/service/DatabaseUpdateService.cpp index 35f44328..98772b79 100644 --- a/service/DatabaseUpdateService.cpp +++ b/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(); -} - - diff --git a/service/DatabaseUpdateService.hpp b/service/DatabaseUpdateService.hpp index 759bb21b..91efc8ca 100644 --- a/service/DatabaseUpdateService.hpp +++ b/service/DatabaseUpdateService.hpp @@ -15,20 +15,15 @@ class DatabaseUpdateService : public Service typedef std::shared_ptr 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 }; diff --git a/service/RemoteServerService.cpp b/service/RemoteServerService.cpp index ddbedabf..6bb6618f 100644 --- a/service/RemoteServerService.cpp +++ b/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 diff --git a/service/RemoteServerService.hpp b/service/RemoteServerService.hpp index 3f70eac9..60d78b57 100644 --- a/service/RemoteServerService.hpp +++ b/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); diff --git a/service/ServiceManager.cpp b/service/ServiceManager.cpp index 67b9dead..7ec84c10 100644 --- a/service/ServiceManager.cpp +++ b/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; } diff --git a/service/ServiceManager.hpp b/service/ServiceManager.hpp index a8212989..cc2d15a2 100644 --- a/service/ServiceManager.hpp +++ b/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 typename T::pointer getService(); boost::mutex& mutex() { return _mutex;}