/src/Fast-DDS/thirdparty/filewatch/FileWatch.hpp

Source (jump to first uncovered line)
//  MIT License
//
//  Copyright(c) 2017 Thomas Monkman
//
//  Permission is hereby granted, free of charge, to any person obtaining a copy
//  of this software and associated documentation files(the "Software"), to deal
//  in the Software without restriction, including without limitation the rights
//  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//  copies of the Software, and to permit persons to whom the Software is
//  furnished to do so, subject to the following conditions :
//
//  The above copyright notice and this permission notice shall be included in all
//  copies or substantial portions of the Software.
//
//  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
//  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
//  SOFTWARE.

#ifndef FILEWATCHER_H
#define FILEWATCHER_H

#include <fastdds/rtps/attributes/ThreadSettings.hpp>

#include <utils/thread.hpp>
#include <utils/threading.hpp>

#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#ifndef MINGW_COMPILER
    #define stat _stat  // do not use stat in windows
#endif  // ifndef MINGW_COMPILER
#ifndef NOMINMAX
#define NOMINMAX
#endif
#include <windows.h>
#include <stdlib.h>
#include <stdio.h>
#include <tchar.h>
#include <Pathcch.h>
#include <shlwapi.h>
#endif // WIN32

#if __unix__
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/inotify.h>
#include <sys/stat.h>
#include <unistd.h>
#endif // __unix__

#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <map>
#include <mutex>
#include <regex>
#include <string>
#include <system_error>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>

namespace eprosima {
namespace filewatch {
    enum class Event {
        added,
        removed,
        modified,
        renamed_old,
        renamed_new
    };

    /**
    * \class FileWatch
    *
    * \brief Watches a folder or file, and will notify of changes via function callback.
    *
    * \author Thomas Monkman
    *
    */
    template<class T>
    class FileWatch
    {
        typedef typename T::value_type C;
        typedef std::basic_string<C, std::char_traits<C>> UnderpinningString;
        typedef std::basic_regex<C, std::regex_traits<C>> UnderpinningRegex;

    public:

        FileWatch(
                T path,
                UnderpinningRegex pattern,
                std::function<void(const T& file, const Event event_type)> callback,
                const fastdds::rtps::ThreadSettings& watch_thread_config,
                const fastdds::rtps::ThreadSettings& callback_thread_config)
            : _path(path)
            , _pattern(pattern)
            , _callback(callback)
            , _directory(get_directory(path))
        {
            init(watch_thread_config, callback_thread_config);
        }

        FileWatch(
                T path,
                std::function<void(const T& file, const Event event_type)> callback,
                const fastdds::rtps::ThreadSettings& watch_thread_config,
                const fastdds::rtps::ThreadSettings& callback_thread_config)
            : FileWatch<T>(path, UnderpinningRegex(_regex_all), callback, watch_thread_config, callback_thread_config) {}

        ~FileWatch() {
            destroy();
        }

        FileWatch(const FileWatch<T>& other) = delete;
        FileWatch<T>& operator=(const FileWatch<T>& other) = delete;

        // Const memeber varibles don't let me implent moves nicely, if moves are really wanted std::unique_ptr should be used and move that.
        FileWatch(FileWatch<T>&&) = delete;
        FileWatch<T>& operator=(FileWatch<T>&&) & = delete;

    private:
        static constexpr C _regex_all[] = { '.', '*', '\0' };
        static constexpr C _this_directory[] = { '.', '/', '\0' };

        struct PathParts
        {
            PathParts(T directory, T filename) : directory(directory), filename(filename) {}
            T directory;
            T filename;
        };
        const T _path;

        UnderpinningRegex _pattern;

        static constexpr std::size_t _buffer_size = { 1024 * 256 };

        // only used if watch a single file
        bool _watching_single_file = { false };
        T _filename;

        std::atomic<bool> _destory = { false };
        std::function<void(const T& file, const Event event_type)> _callback;

        eprosima::thread _watch_thread;

        std::condition_variable _cv;
        std::mutex _callback_mutex;
        std::vector<std::pair<T, Event>> _callback_information;
        eprosima::thread _callback_thread;

        std::promise<void> _running;

        std::chrono::time_point<std::chrono::system_clock> last_write_time_;
        unsigned long last_size_;

#ifdef _WIN32
        HANDLE _directory = { nullptr };
        HANDLE _close_event = { nullptr };

        const DWORD _listen_filters =
            FILE_NOTIFY_CHANGE_SECURITY |
            FILE_NOTIFY_CHANGE_CREATION |
            FILE_NOTIFY_CHANGE_LAST_ACCESS |
            FILE_NOTIFY_CHANGE_LAST_WRITE |
            FILE_NOTIFY_CHANGE_SIZE |
            FILE_NOTIFY_CHANGE_ATTRIBUTES |
            FILE_NOTIFY_CHANGE_DIR_NAME |
            FILE_NOTIFY_CHANGE_FILE_NAME;

        const std::map<DWORD, Event> _event_type_mapping = {
            { FILE_ACTION_ADDED, Event::added },
            { FILE_ACTION_REMOVED, Event::removed },
            { FILE_ACTION_MODIFIED, Event::modified },
            { FILE_ACTION_RENAMED_OLD_NAME, Event::renamed_old },
            { FILE_ACTION_RENAMED_NEW_NAME, Event::renamed_new }
        };

        // time epoch translation
        std::pair<ULARGE_INTEGER, std::chrono::time_point<std::chrono::system_clock>> base_;
#endif // WIN32

#if __unix__
        struct FolderInfo {
            int folder;
            int watch;
        };

        FolderInfo  _directory;

        const std::uint32_t _listen_filters = IN_MODIFY | IN_CREATE | IN_DELETE;

        const static std::size_t event_size = (sizeof(struct inotify_event));
#endif // __unix__

        void init(
            const fastdds::rtps::ThreadSettings& watch_thread_config = {},
            const fastdds::rtps::ThreadSettings& callback_thread_config = {})
        {
#ifdef _WIN32
            _close_event = CreateEvent(NULL, TRUE, FALSE, NULL);
            if (!_close_event) {
                throw std::system_error(GetLastError(), std::system_category());
            }
#endif // WIN32
            _callback_thread = create_thread([this]() {
                try {
                    callback_thread();
                } catch (...) {
                    try {
                        _running.set_exception(std::current_exception());
                    }
                    catch (...) {} // set_exception() may throw too
                }
            }, callback_thread_config, "dds.fwatch.cb");
            _watch_thread = create_thread([this]() {
                try {
                    monitor_directory();
                } catch (...) {
                    try {
                        _running.set_exception(std::current_exception());
                    }
                    catch (...) {} // set_exception() may throw too
                }
            }, watch_thread_config, "dds.fwatch");

            std::future<void> future = _running.get_future();
            future.get(); //block until the monitor_directory is up and running
        }

        void destroy()
        {
            _destory = true;
            _running = std::promise<void>();
#ifdef _WIN32
            SetEvent(_close_event);
#elif __unix__
            inotify_rm_watch(_directory.folder, _directory.watch);
#endif // __unix__
            _cv.notify_all();
            _watch_thread.join();
            _callback_thread.join();
#ifdef _WIN32
            CloseHandle(_directory);
#elif __unix__
            close(_directory.folder);
#endif // __unix__
        }

        const PathParts split_directory_and_file(const T& path) const
        {
            const auto predict = [](C character) {
#ifdef _WIN32
                return character == C('\\') || character == C('/');
#elif __unix__
                return character == C('/');
#endif // __unix__
            };

            UnderpinningString path_string = path;
            const auto pivot = std::find_if(path_string.rbegin(), path_string.rend(), predict).base();
            //if the path is something like "test.txt" there will be no directory part, however we still need one, so insert './'
            const T directory = [&]() {
                const auto extracted_directory = UnderpinningString(path_string.begin(), pivot);
                return (extracted_directory.size() > 0) ? extracted_directory : UnderpinningString(_this_directory);
            }();
            const T filename = UnderpinningString(pivot, path_string.end());
            return PathParts(directory, filename);
        }

        bool pass_filter(const UnderpinningString& file_path)
        {
            if (_watching_single_file) {
                const UnderpinningString extracted_filename = { split_directory_and_file(file_path).filename };
                //if we are watching a single file, only that file should trigger action
                return extracted_filename == _filename;
            }
            return std::regex_match(file_path, _pattern);
        }

#ifdef _WIN32
        template<typename... Args> DWORD GetFileAttributesX(const char* lpFileName, Args... args) {
            return GetFileAttributesA(lpFileName, args...);
        }
        template<typename... Args> DWORD GetFileAttributesX(const wchar_t* lpFileName, Args... args) {
            return GetFileAttributesW(lpFileName, args...);
        }

        template<typename... Args> HANDLE CreateFileX(const char* lpFileName, Args... args) {
            return CreateFileA(lpFileName, args...);
        }
        template<typename... Args> HANDLE CreateFileX(const wchar_t* lpFileName, Args... args) {
            return CreateFileW(lpFileName, args...);
        }

        HANDLE get_directory(const T& path)
        {
            auto file_info = GetFileAttributesX(path.c_str());

            if (file_info == INVALID_FILE_ATTRIBUTES)
            {
                throw std::system_error(GetLastError(), std::system_category());
            }
            _watching_single_file = (file_info & FILE_ATTRIBUTE_DIRECTORY) == false;

            const T watch_path = [this, &path]() {
                if (_watching_single_file)
                {
                    const auto parsed_path = split_directory_and_file(path);
                    _filename = parsed_path.filename;
                    return parsed_path.directory;
                }
                else
                {
                    return path;
                }
            }();

            HANDLE directory = CreateFileX(
                watch_path.c_str(),           // pointer to the file name
                FILE_LIST_DIRECTORY,    // access (read/write) mode
                FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, // share mode
                nullptr, // security descriptor
                OPEN_EXISTING,         // how to create
                FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OVERLAPPED, // file attributes
                HANDLE(0));                 // file with attributes to copy

            if (directory == INVALID_HANDLE_VALUE)
            {
                throw std::system_error(GetLastError(), std::system_category());
            }

            init_last_write_time();

            return directory;
        }

        void convert_wstring(const std::wstring& wstr, std::string& out)
        {
            int size_needed = WideCharToMultiByte(CP_UTF8, 0, &wstr[0], (int)wstr.size(), NULL, 0, NULL, NULL);
            out.resize(size_needed, '\0');
            WideCharToMultiByte(CP_UTF8, 0, &wstr[0], (int)wstr.size(), &out[0], size_needed, NULL, NULL);
        }

        void convert_wstring(const std::wstring& wstr, std::wstring& out)
        {
            out = wstr;
        }

        void monitor_directory()
        {
            std::vector<BYTE> buffer(_buffer_size);
            DWORD bytes_returned = 0;
            OVERLAPPED overlapped_buffer{ 0 };

            overlapped_buffer.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
            if (!overlapped_buffer.hEvent) {
                std::cerr << "Error creating monitor event" << std::endl;
            }

            std::array<HANDLE, 2> handles{ overlapped_buffer.hEvent, _close_event };

            auto async_pending = false;
            _running.set_value();
            do {
                std::vector<std::pair<T, Event>> parsed_information;
                ReadDirectoryChangesW(
                    _directory,
                    buffer.data(), static_cast<DWORD>(buffer.size()),
                    TRUE,
                    _listen_filters,
                    &bytes_returned,
                    &overlapped_buffer, NULL);

                async_pending = true;

                switch (WaitForMultipleObjects(2, handles.data(), FALSE, INFINITE))
                {
                case WAIT_OBJECT_0:
                {
                    if (!GetOverlappedResult(_directory, &overlapped_buffer, &bytes_returned, TRUE)) {
                        throw std::system_error(GetLastError(), std::system_category());
                    }
                    async_pending = false;

                    // Get current time
                    _WIN32_FILE_ATTRIBUTE_DATA att;
                    GetFileAttributesExA(_path.c_str(), GetFileExInfoStandard, &att);

                    unsigned long current_size = att.nFileSizeLow;
                    auto current_time = base_.second
                        + std::chrono::duration<
                                typename std::chrono::time_point<std::chrono::system_clock>::rep,
                                std::ratio_multiply<std::hecto, typename std::chrono::nanoseconds::period>>(
                                    reinterpret_cast<ULARGE_INTEGER*>(&att.ftLastWriteTime)->QuadPart - base_.first.QuadPart);

                    if (bytes_returned == 0 || ((current_time == last_write_time_) && current_size == last_size_ )) {
                        break;
                    }

                    FILE_NOTIFY_INFORMATION *file_information = reinterpret_cast<FILE_NOTIFY_INFORMATION*>(&buffer[0]);
                    do
                    {
                        std::wstring changed_file_w{ file_information->FileName, file_information->FileNameLength / sizeof(file_information->FileName[0]) };
                        UnderpinningString changed_file;
                        convert_wstring(changed_file_w, changed_file);
                        if (pass_filter(changed_file))
                        {
                            parsed_information.emplace_back(T{ changed_file }, _event_type_mapping.at(file_information->Action));
                        }

                        last_write_time_ = current_time;
                        last_size_ = current_size;

                        if (file_information->NextEntryOffset == 0) {
                            break;
                        }

                        file_information = reinterpret_cast<FILE_NOTIFY_INFORMATION*>(reinterpret_cast<BYTE*>(file_information) + file_information->NextEntryOffset);
                    } while (true);
                    break;
                }
                case WAIT_OBJECT_0 + 1:
                    // quit
                    break;
                case WAIT_FAILED:
                    break;
                }
                //dispatch callbacks
                {
                    std::lock_guard<std::mutex> lock(_callback_mutex);
                    _callback_information.insert(_callback_information.end(), parsed_information.begin(), parsed_information.end());
                }
                _cv.notify_all();
            } while (_destory == false);

            if (async_pending)
            {
                //clean up running async io
                CancelIo(_directory);
                GetOverlappedResult(_directory, &overlapped_buffer, &bytes_returned, TRUE);
            }
        }
#endif // WIN32

#if __unix__

        bool is_file(const T& path) const
        {
            struct stat statbuf = {};
            if (stat(path.c_str(), &statbuf) != 0)
            {
                throw std::system_error(errno, std::system_category());
            }
            return S_ISREG(statbuf.st_mode);
        }

        FolderInfo get_directory(const T& path)
        {
            const auto folder = inotify_init();
            if (folder < 0)
            {
                throw std::system_error(errno, std::system_category());
            }
            //const auto listen_filters = _listen_filters;

            _watching_single_file = is_file(path);

            const T watch_path = [this, &path]() {
                if (_watching_single_file)
                {
                    const auto parsed_path = split_directory_and_file(path);
                    _filename = parsed_path.filename;
                    return parsed_path.directory;
                }
                else
                {
                    return path;
                }
            }();

            const auto watch = inotify_add_watch(folder, watch_path.c_str(), IN_MODIFY | IN_CREATE | IN_DELETE );
            if (watch < 0)
            {
                throw std::system_error(errno, std::system_category());
            }

            init_last_write_time();

            return { folder, watch };
        }

        void monitor_directory()
        {
            std::vector<char> buffer(_buffer_size);

            _running.set_value();
            while (_destory == false)
            {
                const auto length = read(_directory.folder, static_cast<void*>(buffer.data()), buffer.size());

                struct stat result;
                stat(_path.c_str(), &result);

                using clock = std::chrono::system_clock;
                using duration = clock::duration;
                std::chrono::time_point<clock> current_time;
                current_time += std::chrono::duration_cast<duration>(std::chrono::seconds(result.st_mtim.tv_sec));
                current_time += std::chrono::duration_cast<duration>(std::chrono::nanoseconds(result.st_mtim.tv_nsec));

                unsigned long current_size = result.st_size;

                if (length > 0 && (current_time != last_write_time_ || current_size != last_size_))
                {
                    int i = 0;
                    last_write_time_ = current_time;
                    last_size_ = current_size;
                    std::vector<std::pair<T, Event>> parsed_information;
                    bool already_modified = false;
                    while (i < length)
                    {
                        struct inotify_event *event = reinterpret_cast<struct inotify_event *>(&buffer[i]); // NOLINT
                        if (event->len)
                        {
                            const UnderpinningString changed_file{ event->name };
                            if (pass_filter(changed_file))
                            {
                                if (event->mask & IN_CREATE)
                                {
                                    parsed_information.emplace_back(T{ changed_file }, Event::added);
                                }
                                else if (event->mask & IN_DELETE)
                                {
                                    parsed_information.emplace_back(T{ changed_file }, Event::removed);
                                }
                                else if (event->mask & IN_MODIFY && !already_modified)
                                {
                                    already_modified = true;
                                    parsed_information.emplace_back(T{ changed_file }, Event::modified);
                                }
                            }
                        }
                        i += event_size + event->len;
                    }
                    //dispatch callbacks
                    {
                        std::lock_guard<std::mutex> lock(_callback_mutex);
                        _callback_information.insert(_callback_information.end(), parsed_information.begin(), parsed_information.end());
                    }
                    _cv.notify_all();
                }
            }
        }
#endif // __unix__

        void callback_thread()
        {
            while (_destory == false) {
                std::unique_lock<std::mutex> lock(_callback_mutex);
                if (_callback_information.empty() && _destory == false) {
                    _cv.wait(lock, [this] { return _callback_information.size() > 0 || _destory; });
                }
                decltype(_callback_information) callback_information = {};
                std::swap(callback_information, _callback_information);
                lock.unlock();

                for (const auto& file : callback_information) {
                    if (_callback) {
                        try
                        {
                            _callback(file.first, file.second);
                        }
                        catch (const std::exception&)
                        {
                        }
                    }
                }
            }
        }

        void init_last_write_time()
        {
#ifdef _WIN32
            // Define epoch reference
            GetSystemTimeAsFileTime((LPFILETIME)&base_.first);
            base_.second = std::chrono::system_clock::now();

            // Initialize last_write_time_
            _WIN32_FILE_ATTRIBUTE_DATA att;
            GetFileAttributesExA(_path.c_str(), GetFileExInfoStandard, &att);

            last_write_time_ = base_.second
                + std::chrono::duration<
                        typename std::chrono::time_point<std::chrono::system_clock>::rep,
                        std::ratio_multiply<std::hecto, typename std::chrono::nanoseconds::period>>(
                            reinterpret_cast<ULARGE_INTEGER*>(&att.ftLastWriteTime)->QuadPart - base_.first.QuadPart);

            // Initialize filesize
            last_size_ = att.nFileSizeLow;
#else
            // Initialize last_write_time_
            struct stat result;
            stat(_path.c_str(), &result);

            using duration = std::chrono::system_clock::duration;
            last_write_time_ += std::chrono::duration_cast<duration>(std::chrono::seconds(result.st_mtim.tv_sec));
            last_write_time_ += std::chrono::duration_cast<duration>(std::chrono::nanoseconds(result.st_mtim.tv_nsec));

            // Initialize filesize
            last_size_ = result.st_size;
#endif
            
        }

    };

    template<class T> constexpr typename FileWatch<T>::C FileWatch<T>::_regex_all[];
    template<class T> constexpr typename FileWatch<T>::C FileWatch<T>::_this_directory[];
}  // namespace filewatch
}  // namespace eprosima
#endif

Coverage Report

Created: 2025-06-13 06:45

Line	Count	Source (jump to first uncovered line)
1		// MIT License
2		//
3		// Copyright(c) 2017 Thomas Monkman
4		//
5		// Permission is hereby granted, free of charge, to any person obtaining a copy
6		// of this software and associated documentation files(the "Software"), to deal
7		// in the Software without restriction, including without limitation the rights
8		// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9		// copies of the Software, and to permit persons to whom the Software is
10		// furnished to do so, subject to the following conditions :
11		//
12		// The above copyright notice and this permission notice shall be included in all
13		// copies or substantial portions of the Software.
14		//
15		// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16		// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17		// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
18		// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19		// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20		// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21		// SOFTWARE.
22
23		#ifndef FILEWATCHER_H
24		#define FILEWATCHER_H
25
26		#include <fastdds/rtps/attributes/ThreadSettings.hpp>
27
28		#include <utils/thread.hpp>
29		#include <utils/threading.hpp>
30
31		#ifdef _WIN32
32		#define WIN32_LEAN_AND_MEAN
33		#ifndef MINGW_COMPILER
34		#define stat _stat // do not use stat in windows
35		#endif // ifndef MINGW_COMPILER
36		#ifndef NOMINMAX
37		#define NOMINMAX
38		#endif
39		#include <windows.h>
40		#include <stdlib.h>
41		#include <stdio.h>
42		#include <tchar.h>
43		#include <Pathcch.h>
44		#include <shlwapi.h>
45		#endif // WIN32
46
47		#if __unix__
48		#include <stdio.h>
49		#include <stdlib.h>
50		#include <errno.h>
51		#include <sys/types.h>
52		#include <sys/inotify.h>
53		#include <sys/stat.h>
54		#include <unistd.h>
55		#endif // __unix__
56
57		#include <algorithm>
58		#include <array>
59		#include <atomic>
60		#include <chrono>
61		#include <condition_variable>
62		#include <functional>
63		#include <future>
64		#include <iostream>
65		#include <map>
66		#include <mutex>
67		#include <regex>
68		#include <string>
69		#include <system_error>
70		#include <thread>
71		#include <type_traits>
72		#include <utility>
73		#include <vector>
74
75		namespace eprosima {
76		namespace filewatch {
77		enum class Event {
78		added,
79		removed,
80		modified,
81		renamed_old,
82		renamed_new
83		};
84
85		/**
86		* \class FileWatch
87		*
88		* \brief Watches a folder or file, and will notify of changes via function callback.
89		*
90		* \author Thomas Monkman
91		*
92		*/
93		template<class T>
94		class FileWatch
95		{
96		typedef typename T::value_type C;
97		typedef std::basic_string<C, std::char_traits<C>> UnderpinningString;
98		typedef std::basic_regex<C, std::regex_traits<C>> UnderpinningRegex;
99
100		public:
101
102		FileWatch(
103		T path,
104		UnderpinningRegex pattern,
105		std::function<void(const T& file, const Event event_type)> callback,
106		const fastdds::rtps::ThreadSettings& watch_thread_config,
107		const fastdds::rtps::ThreadSettings& callback_thread_config)
108	0	: _path(path)
109	0	, _pattern(pattern)
110	0	, _callback(callback)
111	0	, _directory(get_directory(path))
112	0	{
113	0	init(watch_thread_config, callback_thread_config);
114	0	}
115
116		FileWatch(
117		T path,
118		std::function<void(const T& file, const Event event_type)> callback,
119		const fastdds::rtps::ThreadSettings& watch_thread_config,
120		const fastdds::rtps::ThreadSettings& callback_thread_config)
121	0	: FileWatch<T>(path, UnderpinningRegex(_regex_all), callback, watch_thread_config, callback_thread_config) {}
122
123	0	~FileWatch() {
124	0	destroy();
125	0	}
126
127		FileWatch(const FileWatch<T>& other) = delete;
128		FileWatch<T>& operator=(const FileWatch<T>& other) = delete;
129
130		// Const memeber varibles don't let me implent moves nicely, if moves are really wanted std::unique_ptr should be used and move that.
131		FileWatch(FileWatch<T>&&) = delete;
132		FileWatch<T>& operator=(FileWatch<T>&&) & = delete;
133
134		private:
135		static constexpr C _regex_all[] = { '.', '*', '\0' };
136		static constexpr C _this_directory[] = { '.', '/', '\0' };
137
138		struct PathParts
139		{
140	0	PathParts(T directory, T filename) : directory(directory), filename(filename) {}
141		T directory;
142		T filename;
143		};
144		const T _path;
145
146		UnderpinningRegex _pattern;
147
148		static constexpr std::size_t _buffer_size = { 1024 * 256 };
149
150		// only used if watch a single file
151		bool _watching_single_file = { false };
152		T _filename;
153
154		std::atomic<bool> _destory = { false };
155		std::function<void(const T& file, const Event event_type)> _callback;
156
157		eprosima::thread _watch_thread;
158
159		std::condition_variable _cv;
160		std::mutex _callback_mutex;
161		std::vector<std::pair<T, Event>> _callback_information;
162		eprosima::thread _callback_thread;
163
164		std::promise<void> _running;
165
166		std::chrono::time_point<std::chrono::system_clock> last_write_time_;
167		unsigned long last_size_;
168
169		#ifdef _WIN32
170		HANDLE _directory = { nullptr };
171		HANDLE _close_event = { nullptr };
172
173		const DWORD _listen_filters =
174		FILE_NOTIFY_CHANGE_SECURITY \|
175		FILE_NOTIFY_CHANGE_CREATION \|
176		FILE_NOTIFY_CHANGE_LAST_ACCESS \|
177		FILE_NOTIFY_CHANGE_LAST_WRITE \|
178		FILE_NOTIFY_CHANGE_SIZE \|
179		FILE_NOTIFY_CHANGE_ATTRIBUTES \|
180		FILE_NOTIFY_CHANGE_DIR_NAME \|
181		FILE_NOTIFY_CHANGE_FILE_NAME;
182
183		const std::map<DWORD, Event> _event_type_mapping = {
184		{ FILE_ACTION_ADDED, Event::added },
185		{ FILE_ACTION_REMOVED, Event::removed },
186		{ FILE_ACTION_MODIFIED, Event::modified },
187		{ FILE_ACTION_RENAMED_OLD_NAME, Event::renamed_old },
188		{ FILE_ACTION_RENAMED_NEW_NAME, Event::renamed_new }
189		};
190
191		// time epoch translation
192		std::pair<ULARGE_INTEGER, std::chrono::time_point<std::chrono::system_clock>> base_;
193		#endif // WIN32
194
195		#if __unix__
196		struct FolderInfo {
197		int folder;
198		int watch;
199		};
200
201		FolderInfo _directory;
202
203		const std::uint32_t _listen_filters = IN_MODIFY \| IN_CREATE \| IN_DELETE;
204
205		const static std::size_t event_size = (sizeof(struct inotify_event));
206		#endif // __unix__
207
208		void init(
209		const fastdds::rtps::ThreadSettings& watch_thread_config = {},
210		const fastdds::rtps::ThreadSettings& callback_thread_config = {})
211	0	{
212		#ifdef _WIN32
213		_close_event = CreateEvent(NULL, TRUE, FALSE, NULL);
214		if (!_close_event) {
215		throw std::system_error(GetLastError(), std::system_category());
216		}
217		#endif // WIN32
218	0	_callback_thread = create_thread([this]() {
219	0	try {
220	0	callback_thread();
221	0	} catch (...) {
222	0	try {
223	0	_running.set_exception(std::current_exception());
224	0	}
225	0	catch (...) {} // set_exception() may throw too
226	0	}
227	0	}, callback_thread_config, "dds.fwatch.cb");
228	0	_watch_thread = create_thread([this]() {
229	0	try {
230	0	monitor_directory();
231	0	} catch (...) {
232	0	try {
233	0	_running.set_exception(std::current_exception());
234	0	}
235	0	catch (...) {} // set_exception() may throw too
236	0	}
237	0	}, watch_thread_config, "dds.fwatch");
238
239	0	std::future<void> future = _running.get_future();
240	0	future.get(); //block until the monitor_directory is up and running
241	0	}
242
243		void destroy()
244	0	{
245	0	_destory = true;
246	0	_running = std::promise<void>();
247		#ifdef _WIN32
248		SetEvent(_close_event);
249		#elif __unix__
250		inotify_rm_watch(_directory.folder, _directory.watch);
251	0	#endif // __unix__
252	0	_cv.notify_all();
253	0	_watch_thread.join();
254	0	_callback_thread.join();
255		#ifdef _WIN32
256		CloseHandle(_directory);
257		#elif __unix__
258		close(_directory.folder);
259	0	#endif // __unix__
260	0	}
261
262		const PathParts split_directory_and_file(const T& path) const
263	0	{
264	0	const auto predict = [](C character) {
265		#ifdef _WIN32
266		return character == C('\\') \|\| character == C('/');
267		#elif __unix__
268		return character == C('/');
269	0	#endif // __unix__
270	0	};
271
272	0	UnderpinningString path_string = path;
273	0	const auto pivot = std::find_if(path_string.rbegin(), path_string.rend(), predict).base();
274		//if the path is something like "test.txt" there will be no directory part, however we still need one, so insert './'
275	0	const T directory = [&]() {
276	0	const auto extracted_directory = UnderpinningString(path_string.begin(), pivot);
277	0	return (extracted_directory.size() > 0) ? extracted_directory : UnderpinningString(_this_directory);
278	0	}();
279	0	const T filename = UnderpinningString(pivot, path_string.end());
280	0	return PathParts(directory, filename);
281	0	}
282
283		bool pass_filter(const UnderpinningString& file_path)
284	0	{
285	0	if (_watching_single_file) {
286	0	const UnderpinningString extracted_filename = { split_directory_and_file(file_path).filename };
287		//if we are watching a single file, only that file should trigger action
288	0	return extracted_filename == _filename;
289	0	}
290	0	return std::regex_match(file_path, _pattern);
291	0	}
292
293		#ifdef _WIN32
294		template<typename... Args> DWORD GetFileAttributesX(const char* lpFileName, Args... args) {
295		return GetFileAttributesA(lpFileName, args...);
296		}
297		template<typename... Args> DWORD GetFileAttributesX(const wchar_t* lpFileName, Args... args) {
298		return GetFileAttributesW(lpFileName, args...);
299		}
300
301		template<typename... Args> HANDLE CreateFileX(const char* lpFileName, Args... args) {
302		return CreateFileA(lpFileName, args...);
303		}
304		template<typename... Args> HANDLE CreateFileX(const wchar_t* lpFileName, Args... args) {
305		return CreateFileW(lpFileName, args...);
306		}
307
308		HANDLE get_directory(const T& path)
309		{
310		auto file_info = GetFileAttributesX(path.c_str());
311
312		if (file_info == INVALID_FILE_ATTRIBUTES)
313		{
314		throw std::system_error(GetLastError(), std::system_category());
315		}
316		_watching_single_file = (file_info & FILE_ATTRIBUTE_DIRECTORY) == false;
317
318		const T watch_path = [this, &path]() {
319		if (_watching_single_file)
320		{
321		const auto parsed_path = split_directory_and_file(path);
322		_filename = parsed_path.filename;
323		return parsed_path.directory;
324		}
325		else
326		{
327		return path;
328		}
329		}();
330
331		HANDLE directory = CreateFileX(
332		watch_path.c_str(), // pointer to the file name
333		FILE_LIST_DIRECTORY, // access (read/write) mode
334		FILE_SHARE_READ \| FILE_SHARE_WRITE \| FILE_SHARE_DELETE, // share mode
335		nullptr, // security descriptor
336		OPEN_EXISTING, // how to create
337		FILE_FLAG_BACKUP_SEMANTICS \| FILE_FLAG_OVERLAPPED, // file attributes
338		HANDLE(0)); // file with attributes to copy
339
340		if (directory == INVALID_HANDLE_VALUE)
341		{
342		throw std::system_error(GetLastError(), std::system_category());
343		}
344
345		init_last_write_time();
346
347		return directory;
348		}
349
350		void convert_wstring(const std::wstring& wstr, std::string& out)
351		{
352		int size_needed = WideCharToMultiByte(CP_UTF8, 0, &wstr[0], (int)wstr.size(), NULL, 0, NULL, NULL);
353		out.resize(size_needed, '\0');
354		WideCharToMultiByte(CP_UTF8, 0, &wstr[0], (int)wstr.size(), &out[0], size_needed, NULL, NULL);
355		}
356
357		void convert_wstring(const std::wstring& wstr, std::wstring& out)
358		{
359		out = wstr;
360		}
361
362		void monitor_directory()
363		{
364		std::vector<BYTE> buffer(_buffer_size);
365		DWORD bytes_returned = 0;
366		OVERLAPPED overlapped_buffer{ 0 };
367
368		overlapped_buffer.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
369		if (!overlapped_buffer.hEvent) {
370		std::cerr << "Error creating monitor event" << std::endl;
371		}
372
373		std::array<HANDLE, 2> handles{ overlapped_buffer.hEvent, _close_event };
374
375		auto async_pending = false;
376		_running.set_value();
377		do {
378		std::vector<std::pair<T, Event>> parsed_information;
379		ReadDirectoryChangesW(
380		_directory,
381		buffer.data(), static_cast<DWORD>(buffer.size()),
382		TRUE,
383		_listen_filters,
384		&bytes_returned,
385		&overlapped_buffer, NULL);
386
387		async_pending = true;
388
389		switch (WaitForMultipleObjects(2, handles.data(), FALSE, INFINITE))
390		{
391		case WAIT_OBJECT_0:
392		{
393		if (!GetOverlappedResult(_directory, &overlapped_buffer, &bytes_returned, TRUE)) {
394		throw std::system_error(GetLastError(), std::system_category());
395		}
396		async_pending = false;
397
398		// Get current time
399		_WIN32_FILE_ATTRIBUTE_DATA att;
400		GetFileAttributesExA(_path.c_str(), GetFileExInfoStandard, &att);
401
402		unsigned long current_size = att.nFileSizeLow;
403		auto current_time = base_.second
404		+ std::chrono::duration<
405		typename std::chrono::time_point<std::chrono::system_clock>::rep,
406		std::ratio_multiply<std::hecto, typename std::chrono::nanoseconds::period>>(
407		reinterpret_cast<ULARGE_INTEGER*>(&att.ftLastWriteTime)->QuadPart - base_.first.QuadPart);
408
409		if (bytes_returned == 0 \|\| ((current_time == last_write_time_) && current_size == last_size_ )) {
410		break;
411		}
412
413		FILE_NOTIFY_INFORMATION file_information = reinterpret_cast<FILE_NOTIFY_INFORMATION>(&buffer[0]);
414		do
415		{
416		std::wstring changed_file_w{ file_information->FileName, file_information->FileNameLength / sizeof(file_information->FileName[0]) };
417		UnderpinningString changed_file;
418		convert_wstring(changed_file_w, changed_file);
419		if (pass_filter(changed_file))
420		{
421		parsed_information.emplace_back(T{ changed_file }, _event_type_mapping.at(file_information->Action));
422		}
423
424		last_write_time_ = current_time;
425		last_size_ = current_size;
426
427		if (file_information->NextEntryOffset == 0) {
428		break;
429		}
430
431		file_information = reinterpret_cast<FILE_NOTIFY_INFORMATION>(reinterpret_cast<BYTE>(file_information) + file_information->NextEntryOffset);
432		} while (true);
433		break;
434		}
435		case WAIT_OBJECT_0 + 1:
436		// quit
437		break;
438		case WAIT_FAILED:
439		break;
440		}
441		//dispatch callbacks
442		{
443		std::lock_guard<std::mutex> lock(_callback_mutex);
444		_callback_information.insert(_callback_information.end(), parsed_information.begin(), parsed_information.end());
445		}
446		_cv.notify_all();
447		} while (_destory == false);
448
449		if (async_pending)
450		{
451		//clean up running async io
452		CancelIo(_directory);
453		GetOverlappedResult(_directory, &overlapped_buffer, &bytes_returned, TRUE);
454		}
455		}
456		#endif // WIN32
457
458		#if __unix__
459
460		bool is_file(const T& path) const
461	0	{
462	0	struct stat statbuf = {};
463	0	if (stat(path.c_str(), &statbuf) != 0)
464	0	{
465	0	throw std::system_error(errno, std::system_category());
466	0	}
467	0	return S_ISREG(statbuf.st_mode);
468	0	}
469
470		FolderInfo get_directory(const T& path)
471	0	{
472	0	const auto folder = inotify_init();
473	0	if (folder < 0)
474	0	{
475	0	throw std::system_error(errno, std::system_category());
476	0	}
477		//const auto listen_filters = _listen_filters;
478
479	0	_watching_single_file = is_file(path);
480
481	0	const T watch_path = [this, &path]() {
482	0	if (_watching_single_file)
483	0	{
484	0	const auto parsed_path = split_directory_and_file(path);
485	0	_filename = parsed_path.filename;
486	0	return parsed_path.directory;
487	0	}
488	0	else
489	0	{
490	0	return path;
491	0	}
492	0	}();
493
494	0	const auto watch = inotify_add_watch(folder, watch_path.c_str(), IN_MODIFY \| IN_CREATE \| IN_DELETE );
495	0	if (watch < 0)
496	0	{
497	0	throw std::system_error(errno, std::system_category());
498	0	}
499
500	0	init_last_write_time();
501
502	0	return { folder, watch };
503	0	}
504
505		void monitor_directory()
506	0	{
507	0	std::vector<char> buffer(_buffer_size);
508
509	0	_running.set_value();
510	0	while (_destory == false)
511	0	{
512	0	const auto length = read(_directory.folder, static_cast<void*>(buffer.data()), buffer.size());
513
514	0	struct stat result;
515	0	stat(_path.c_str(), &result);
516
517	0	using clock = std::chrono::system_clock;
518	0	using duration = clock::duration;
519	0	std::chrono::time_point<clock> current_time;
520	0	current_time += std::chrono::duration_cast<duration>(std::chrono::seconds(result.st_mtim.tv_sec));
521	0	current_time += std::chrono::duration_cast<duration>(std::chrono::nanoseconds(result.st_mtim.tv_nsec));
522
523	0	unsigned long current_size = result.st_size;
524
525	0	if (length > 0 && (current_time != last_write_time_ \|\| current_size != last_size_))
526	0	{
527	0	int i = 0;
528	0	last_write_time_ = current_time;
529	0	last_size_ = current_size;
530	0	std::vector<std::pair<T, Event>> parsed_information;
531	0	bool already_modified = false;
532	0	while (i < length)
533	0	{
534	0	struct inotify_event event = reinterpret_cast<struct inotify_event >(&buffer[i]); // NOLINT
535	0	if (event->len)
536	0	{
537	0	const UnderpinningString changed_file{ event->name };
538	0	if (pass_filter(changed_file))
539	0	{
540	0	if (event->mask & IN_CREATE)
541	0	{
542	0	parsed_information.emplace_back(T{ changed_file }, Event::added);
543	0	}
544	0	else if (event->mask & IN_DELETE)
545	0	{
546	0	parsed_information.emplace_back(T{ changed_file }, Event::removed);
547	0	}
548	0	else if (event->mask & IN_MODIFY && !already_modified)
549	0	{
550	0	already_modified = true;
551	0	parsed_information.emplace_back(T{ changed_file }, Event::modified);
552	0	}
553	0	}
554	0	}
555	0	i += event_size + event->len;
556	0	}
557		//dispatch callbacks
558	0	{
559	0	std::lock_guard<std::mutex> lock(_callback_mutex);
560	0	_callback_information.insert(_callback_information.end(), parsed_information.begin(), parsed_information.end());
561	0	}
562	0	_cv.notify_all();
563	0	}
564	0	}
565	0	}
566		#endif // __unix__
567
568		void callback_thread()
569	0	{
570	0	while (_destory == false) {
571	0	std::unique_lock<std::mutex> lock(_callback_mutex);
572	0	if (_callback_information.empty() && _destory == false) {
573	0	_cv.wait(lock, [this] { return _callback_information.size() > 0 \|\| _destory; });
574	0	}
575	0	decltype(_callback_information) callback_information = {};
576	0	std::swap(callback_information, _callback_information);
577	0	lock.unlock();
578
579	0	for (const auto& file : callback_information) {
580	0	if (_callback) {
581	0	try
582	0	{
583	0	_callback(file.first, file.second);
584	0	}
585	0	catch (const std::exception&)
586	0	{
587	0	}
588	0	}
589	0	}
590	0	}
591	0	}
592
593		void init_last_write_time()
594	0	{
595		#ifdef _WIN32
596		// Define epoch reference
597		GetSystemTimeAsFileTime((LPFILETIME)&base_.first);
598		base_.second = std::chrono::system_clock::now();
599
600		// Initialize last_write_time_
601		_WIN32_FILE_ATTRIBUTE_DATA att;
602		GetFileAttributesExA(_path.c_str(), GetFileExInfoStandard, &att);
603
604		last_write_time_ = base_.second
605		+ std::chrono::duration<
606		typename std::chrono::time_point<std::chrono::system_clock>::rep,
607		std::ratio_multiply<std::hecto, typename std::chrono::nanoseconds::period>>(
608		reinterpret_cast<ULARGE_INTEGER*>(&att.ftLastWriteTime)->QuadPart - base_.first.QuadPart);
609
610		// Initialize filesize
611		last_size_ = att.nFileSizeLow;
612		#else
613		// Initialize last_write_time_
614	0	struct stat result;
615	0	stat(_path.c_str(), &result);
616
617	0	using duration = std::chrono::system_clock::duration;
618	0	last_write_time_ += std::chrono::duration_cast<duration>(std::chrono::seconds(result.st_mtim.tv_sec));
619	0	last_write_time_ += std::chrono::duration_cast<duration>(std::chrono::nanoseconds(result.st_mtim.tv_nsec));
620
621		// Initialize filesize
622	0	last_size_ = result.st_size;
623	0	#endif
624
625	0	}
626
627		};
628
629		template<class T> constexpr typename FileWatch<T>::C FileWatch<T>::_regex_all[];
630		template<class T> constexpr typename FileWatch<T>::C FileWatch<T>::_this_directory[];
631		} // namespace filewatch
632		} // namespace eprosima
633		#endif