Old sample using inotify with ASIO

inotify_with_asio_sample.cpp

// My old PoC for notify (just for memo).
// Sample based on https://lwn.net/Articles/604686/
//                 http://think-async.com/Asio/asio-1.11.0/doc/asio/reference/posix__stream_descriptor.html

#include <sys/inotify.h>

#include <iostream>
#include <thread>
#include <stdexcept>
#include <memory>
#include <map>

#include <asio.hpp>

//
// io_service_wrapper_t
//

//! An io_service wrapper class.
class io_service_wrapper_t
{
		io_service_wrapper_t( const io_service_wrapper_t & );
		void
		operator = ( const io_service_wrapper_t & );

	public:
		io_service_wrapper_t()
		{}

		~io_service_wrapper_t()
		{
			shutdown();
			wait();
		}

		//! Run service iface.
		//! \{
		void
		start();

		void
		shutdown();

		void
		wait();
		//! \}

		asio::io_service &
		io_service()
		{
			return m_io_service;
		}

	private:
		//! A thread to run an io_service.
		std::thread m_runner;

		//! Target io_service instance.
		asio::io_service m_io_service;

		//! A fake work to run io_service as an endless loop.
		std::unique_ptr< asio::io_service::work > m_ensure_run_work;
};

void
io_service_wrapper_t::start()
{
	m_runner =
		std::thread( [this ](){
			try
			{
				// Ensure that the io_service run() function will not exit
				// until stop() is called.
				m_ensure_run_work.reset(
					new asio::io_service::work( m_io_service ) );

				// start working circle.
				m_io_service.run();
			}
			catch( const std::exception & ex )
			{
				std::cerr
					<< "Unexpected error while running asio::ioservice: "
					<< ex.what();
			}
		} );
}

void
io_service_wrapper_t::shutdown()
{
	m_io_service.post(
		[this ]()
		{
			m_ensure_run_work.reset();
		} );
}

void
io_service_wrapper_t::wait()
{
	m_runner.join();
}

typedef std::function< void( const std::string & ) >
	file_created_handler_t;

//
// watch_descriptor_wrapper_t
//

//! A single dir watcher wrapper.
struct watch_descriptor_wrapper_t
	:
		public std::enable_shared_from_this< watch_descriptor_wrapper_t >
{
	watch_descriptor_wrapper_t(
		int inotify_fd,
		int inotify_wd,
		file_created_handler_t handler )
		:
			m_inotify_fd( inotify_fd ),
			m_inotify_wd( inotify_wd ),
			m_handler( std::move( handler ) )
	{}

	~watch_descriptor_wrapper_t()
	{
		inotify_rm_watch( m_inotify_fd, m_inotify_wd );
	}

	//! inotify descriptors.
	//! \{
	int m_inotify_fd;
	int m_inotify_wd;
	//! \}

	file_created_handler_t m_handler;
};

typedef std::shared_ptr< watch_descriptor_wrapper_t >
	watch_descriptor_wrapper_shared_ptr_t;


//
// inotify_wrapper_t
//

//! Sort of top-level interface wrapper for working with inotify.
class inotify_wrapper_t
	:
		public std::enable_shared_from_this< inotify_wrapper_t >
{
	public:
		inotify_wrapper_t(
			asio::io_service & io_service );

		~inotify_wrapper_t();

		void
		start_watch_dir(
			const std::string & path,
			file_created_handler_t file_created_handler );

		void
		stop_watch_dir(
			const std::string & path );

		void
		start();

		void
		stop();

	private:
		asio::posix::stream_descriptor m_stream_descriptor;

		int m_inotify_fd{ -1 };

		typedef std::map< int, watch_descriptor_wrapper_shared_ptr_t >
			watchers_table_t;

		//! What we watch currently.
		watchers_table_t m_watchers;

		typedef std::map< std::string, int > path_to_wd_table_t;

		//! Dir-path => watch descriptors.
		path_to_wd_table_t m_path_to_wds;

		std::array< char, 8 * 1024 > m_buf;
};

inotify_wrapper_t::inotify_wrapper_t(
	asio::io_service & io_service )
	:
		m_stream_descriptor{ io_service }
{
	m_inotify_fd = inotify_init1( IN_NONBLOCK );
	if( -1 == m_inotify_fd )
	{
		throw std::runtime_error( "inotify_init() failed" );
	}

	m_stream_descriptor =
		asio::posix::stream_descriptor{ io_service, m_inotify_fd };
}

inotify_wrapper_t::~inotify_wrapper_t()
{
	m_stream_descriptor.release();

	m_watchers.clear();
	m_path_to_wds.clear();

	if( -1 == m_inotify_fd )
		close( m_inotify_fd );
}


void
inotify_wrapper_t::start_watch_dir(
	const std::string & path,
	file_created_handler_t file_created_handler )
{
	if( m_path_to_wds.end() != m_path_to_wds.find( path ) )
		throw std::runtime_error( "'" + path + "' is already watched" );

	int inotify_wd =
		inotify_add_watch(
			m_inotify_fd,
			path.c_str(),
			IN_CLOSE_WRITE | IN_MOVED_TO | IN_ONLYDIR );

	if( -1 == inotify_wd )
		throw std::runtime_error( "unable to watch " + path );

	try
	{
		auto h =
			std::make_shared< watch_descriptor_wrapper_t >(
				m_inotify_fd,
				inotify_wd,
				std::move( file_created_handler ) );

		m_path_to_wds[ path ] = inotify_wd;

		m_watchers[ inotify_wd ] = h;
	}
	catch( ... )
	{
		m_path_to_wds.erase( path );
		m_watchers.erase( inotify_wd );
	}
}

void
inotify_wrapper_t::stop_watch_dir(
	const std::string & path )
{
	auto it = m_path_to_wds.find( path );

	if( m_path_to_wds.end() == it )
	{
		m_watchers.erase( it->second );
		m_path_to_wds.erase( it );
	}
}

void
inotify_wrapper_t::start()
{
	auto self = shared_from_this();

	m_stream_descriptor.async_read_some(
		asio::buffer(
			m_buf.data(),
			m_buf.size() ),
		[ self, this ](
			const asio::error_code& ec,
			std::size_t len ){
			if( asio::error::operation_aborted == ec.value() )
			{
				return;
			}

			if( !ec )
			{
				for( char * p = m_buf.data(); p < m_buf.data() + len; )
				{
					try
					{
						inotify_event * event = (inotify_event *)p;

						auto it = m_watchers.find( event->wd );

						if( m_watchers.end() != it )
							it->second->m_handler( event->name );

						p += sizeof( struct inotify_event ) + event->len;
					}
					catch( const std::exception & ex )
					{
						std::cout << "error: " << ex.what() << std::endl;
					}
				}
			}

			start();
		} );
}

void
inotify_wrapper_t::stop()
{
	m_stream_descriptor.cancel();
}

int
main(int argc, char* argv[])
{
	try
	{
		if (argc < 2)
		{
			std::cerr << "Usage: dirwatcher <path1> <path2> ..." << std::endl;
			return 1;
		}

		std::vector< std::string > dirs{ argv +1, argv + argc };

		io_service_wrapper_t io_service_wrapper;
		io_service_wrapper.start();

		auto inotify_wrapper =
			std::make_shared< inotify_wrapper_t >(
				io_service_wrapper.io_service() );

		asio::post(
			io_service_wrapper.io_service().get_executor(),
			[ inotify_wrapper, dirs ](){
				for( const auto & d : dirs )
				{
					inotify_wrapper->start_watch_dir(
						d,
						[ d ]( const std::string & p ){
							std::cout
								<< "trigger on " << d << ": "
								<< p << std::endl;
						} );
				}

				inotify_wrapper->start();
			} );

		std::cout << "Enter quit or q to exit..." << std::endl;

		while( true )
		{
			std::string input;

			std::cin >> input;

			if( input == "q" || input == "quit" )
				break;
		}


		asio::post(
			io_service_wrapper.io_service().get_executor(),
			[ inotify_wrapper ](){
				inotify_wrapper->stop();
			} );

	}
	catch( const std::exception & x )
	{
		std::cerr << "Exception: " << x.what() << std::endl;
		return 2;
	}

	return 0;
}