process inotify events

common/cpp/include/common/data_structs.h

@@ -31,6 +31,7 @@ inline bool operator==(const FileInfo& lhs, const FileInfo& rhs) {
 
 using FileData = std::unique_ptr<uint8_t[]>;
 using FileInfos = std::vector<FileInfo>;
+using SubDirList = std::vector<std::string>;
 
 }
 #endif //ASAPO_FILE_INFO_H

common/cpp/include/io/io.h

@@ -96,8 +96,7 @@ class IO {
 
     virtual void            CreateNewDirectory      (const std::string& directory_name, Error* err) const = 0;
     virtual FileData        GetDataFromFile         (const std::string& fname, uint64_t fsize, Error* err) const = 0;
-    virtual void CollectFileInformationRecursively(const std::string& path, std::vector<FileInfo>* files,
-                                                   Error* err) const = 0;
+    virtual SubDirList      GetSubDirectories(const std::string& path, Error* err) const = 0;
     virtual std::vector<FileInfo>   FilesInFolder   (const std::string& folder, Error* err) const = 0;
     virtual std::string     ReadFileToString        (const std::string& fname, Error* err) const = 0;
 

common/cpp/include/unittests/MockIO.h

@@ -185,15 +185,6 @@ class MockIO : public IO {
 
     MOCK_CONST_METHOD3(WriteDataToFile_t, ErrorInterface * (const std::string& fname, const uint8_t* data, size_t fsize));
 
-    void CollectFileInformationRecursively(const std::string& path, std::vector<FileInfo>* files,
-                                           Error* err) const override {
-        ErrorInterface* error = nullptr;
-        CollectFileInformationRecursivly_t(path, files, &error);
-        err->reset(error);
-    }
-    MOCK_CONST_METHOD3(CollectFileInformationRecursivly_t, void(const std::string& path, std::vector<FileInfo>* files,
-                       ErrorInterface** err));
-
     std::vector<FileInfo> FilesInFolder(const std::string& folder, Error* err) const override {
         ErrorInterface* error = nullptr;
         auto data = FilesInFolder_t(folder, &error);
@@ -202,6 +193,17 @@ class MockIO : public IO {
     }
     MOCK_CONST_METHOD2(FilesInFolder_t, std::vector<FileInfo>(const std::string& folder, ErrorInterface** err));
 
+
+    SubDirList GetSubDirectories(const std::string& path, Error* err) const override {
+        ErrorInterface* error = nullptr;
+        auto data = GetSubDirectories_t(path, &error);
+        err->reset(error);
+        return data;
+    };
+
+    MOCK_CONST_METHOD2(GetSubDirectories_t, SubDirList(const std::string& path, ErrorInterface** err));
+
+
     std::string ReadFileToString(const std::string& fname, Error* err) const override {
         ErrorInterface* error = nullptr;
         auto data = ReadFileToString_t(fname, &error);

common/cpp/src/system_io/system_io.cpp

@@ -114,6 +114,16 @@ FileInfos SystemIO::FilesInFolder(const std::string& folder, Error* err) const {
     return files;
 }
 
+SubDirList SystemIO::GetSubDirectories(const std::string& path, Error* err) const {
+    SubDirList list;
+    GetSubDirectoriesRecursively(path, &list, err);
+    if (*err != nullptr) {
+        return {};
+    }
+    return list;
+}
+
+
 void asapo::SystemIO::CreateNewDirectory(const std::string& directory_name, Error* err) const {
     if(_mkdir(directory_name.c_str()) == -1) {
         *err = GetLastError();

common/cpp/src/system_io/system_io.h

@@ -61,7 +61,9 @@ class SystemIO final : public IO {
     static ssize_t		_recv(SocketDescriptor socket_fd, void* buffer, size_t length);
     static ssize_t      _read(FileDescriptor fd, void* buffer, size_t length);
     static ssize_t      _write(FileDescriptor fd, const void* buffer, size_t count);
-
+    void            CollectFileInformationRecursively(const std::string& path, std::vector<FileInfo>* files,
+                                                      Error* err) const;
+    void            GetSubDirectoriesRecursively(const std::string& path, SubDirList* subdirs, Error* err) const;
   public:
     /*
      * Special
@@ -95,17 +97,16 @@ class SystemIO final : public IO {
     /*
      * Filesystem
      */
-    FileDescriptor  Open(const std::string& filename, int open_flags, Error* err) const;
-    void            Close(FileDescriptor fd, Error* err) const;
-    size_t          Read(FileDescriptor fd, void* buf, size_t length, Error* err) const;
-    size_t          Write(FileDescriptor fd, const void* buf, size_t length, Error* err) const;
-    void            CreateNewDirectory(const std::string& directory_name, Error* err) const;
-    FileData        GetDataFromFile(const std::string& fname, uint64_t fsize, Error* err) const;
-    Error           WriteDataToFile  (const std::string& fname, const FileData& data, size_t length) const;
-    Error           WriteDataToFile(const std::string& fname, const uint8_t* data, size_t length) const;
-    void            CollectFileInformationRecursively(const std::string& path, std::vector<FileInfo>* files,
-                                                      Error* err) const;
-    std::string     ReadFileToString(const std::string& fname, Error* err) const;
+    FileDescriptor  Open(const std::string& filename, int open_flags, Error* err) const override;
+    void            Close(FileDescriptor fd, Error* err) const override;
+    size_t          Read(FileDescriptor fd, void* buf, size_t length, Error* err) const override;
+    size_t          Write(FileDescriptor fd, const void* buf, size_t length, Error* err) const override;
+    void            CreateNewDirectory(const std::string& directory_name, Error* err) const override;
+    FileData        GetDataFromFile(const std::string& fname, uint64_t fsize, Error* err) const override;
+    Error           WriteDataToFile  (const std::string& fname, const FileData& data, size_t length) const override;
+    Error           WriteDataToFile(const std::string& fname, const uint8_t* data, size_t length) const override;
+    SubDirList      GetSubDirectories(const std::string& path, Error* err) const override;
+    std::string     ReadFileToString(const std::string& fname, Error* err) const override;
 };
 }
 

common/cpp/src/system_io/system_io_linux.cpp

@@ -145,7 +145,32 @@ void ProcessFileEntity(const struct dirent* entity, const std::string& path,
     files->push_back(file_info);
 }
 
-/** @} */
+
+
+void SystemIO::GetSubDirectoriesRecursively(const std::string& path, SubDirList* subdirs, Error* err) const {
+    errno = 0;
+    auto dir = opendir((path).c_str());
+    if (dir == nullptr) {
+        *err = GetLastError();
+        (*err)->Append(path);
+        return;
+    }
+
+    while (struct dirent* current_entity = readdir(dir)) {
+        if (IsDirectory(current_entity)) {
+            std::string subdir = path + "/" + current_entity->d_name;
+            subdirs->push_back(subdir);
+            GetSubDirectoriesRecursively(subdir, subdirs, err);
+        }
+        if (*err != nullptr) {
+            errno = 0;
+            closedir(dir);
+            return;
+        }
+    }
+    *err = GetLastError();
+    closedir(dir);
+}
 
 void SystemIO::CollectFileInformationRecursively(const std::string& path,
                                                  FileInfos* files, Error* err)  const {

common/cpp/src/system_io/system_io_windows.cpp

@@ -154,6 +154,35 @@ void ProcessFileEntity(const WIN32_FIND_DATA& f, const std::string& path,
     files->push_back(file_info);
 }
 
+void GetSubDirectoriesRecursively(const std::string& path, SubDirList* subdirs, Error* err) const {
+    WIN32_FIND_DATA find_data;
+    HANDLE handle = FindFirstFile((path + "\\*.*").c_str(), &find_data);
+    if (handle == INVALID_HANDLE_VALUE) {
+        *err = IOErrorFromGetLastError();
+        (*err)->Append(path);
+        return;
+    }
+
+    do {
+        if (IsDirectory(find_data)) {
+            std::string subdir = path + "\\" + find_data.cFileName;
+            subdirs->push_back(subdir);
+            GetSubDirectoriesRecursively(subdir, subdirs, err);
+        }
+        if (*err) {
+            FindClose(handle);
+            return;
+        }
+    } while (FindNextFile(handle, &find_data));
+
+    if (FindClose(handle)) {
+        *err = nullptr;
+    } else {
+        *err = IOErrorFromGetLastError();
+    }
+}
+
+
 void SystemIO::CollectFileInformationRecursively(const std::string& path,
                                                  FileInfos* files, Error* err) const {
     WIN32_FIND_DATA find_data;

producer/event_monitor_producer/src/event_monitor_error.h

@@ -37,6 +37,11 @@ class EventMonitorErrorTemplate : public SimpleErrorTemplate {
         return error_type_;
     }
 
+    inline Error Generate(std::string sub_error) const noexcept {
+        return Error(new EventMonitorError(error_ + ": " + sub_error, error_type_));
+    }
+
+
     inline Error Generate() const noexcept override {
         return Error(new EventMonitorError(error_, error_type_));
     }

producer/event_monitor_producer/src/eventmon_main.cpp

@@ -69,7 +69,7 @@ int main (int argc, char* argv[]) {
     stop_signal = 0;
     std::signal(SIGINT, SignalHandler);
     std::signal(SIGTERM, SignalHandler);
-
+    siginterrupt(SIGINT, 1);
 
     const auto& logger = asapo::GetDefaultEventMonLogger();
     logger->SetLogLevel(GetEventMonConfig()->log_level);
@@ -81,13 +81,17 @@ int main (int argc, char* argv[]) {
 
     err = event_detector->StartMonitoring();
     if (err) {
+        logger->Error(err->Explain());
         return EXIT_FAILURE;
     }
 
     int i = 0;
-    while (!stop_signal) {
+    while (true) {
         asapo::EventHeader event_header;
         auto err = event_detector->GetNextEvent(&event_header);
+        if (stop_signal) {
+            break; // we check it here because signal can interrupt system call (ready by inotify and result n incomplete event data)
+        }
         if (err) {
             if (err != asapo::EventMonitorErrorTemplates::kNoNewEvent) {
                 logger->Error("cannot retrieve next event: " + err->Explain());

producer/event_monitor_producer/src/system_folder_watch_linux.cpp

 #include "system_folder_watch_linux.h"
 
+
+#include "event_monitor_error.h"
+#include "eventmon_logger.h"
+
 namespace asapo {
 
+Error SystemFolderWatch::AddFolderToWatch(std::string folder, bool recursive) {
+    int id = inotify_add_watch(watch_fd_, folder.c_str(),
+                               IN_CLOSE_WRITE
+                               | IN_MOVED_TO
+                               | IN_MOVED_FROM
+                               | IN_CREATE
+                               | IN_DELETE_SELF
+//                               | IN_MOVE_SELF
+                               | IN_EXCL_UNLINK
+                               | IN_DONT_FOLLOW
+                               | IN_ONLYDIR);
+    if (id == -1) {
+        return EventMonitorErrorTemplates::kSystemError.Generate("cannot add watch for " + folder);
+    } else {
+        GetDefaultEventMonLogger()->Debug("added folder to monitor: " + folder);
+    }
+    watched_folders_paths_[id] = folder;
+    if (recursive) {
+        Error err;
+        auto subdirs = io_-> GetSubDirectories(folder, &err);
+        if (err) {
+            return err;
+        }
+        for (auto& subdir : subdirs) {
+            err = AddFolderToWatch(subdir, false);
+            if (err) {
+                return err;
+            }
+        }
+
+    }
+    return nullptr;
+}
+
+
 Error SystemFolderWatch::StartFolderMonitor(const std::vector<std::string>& monitored_folders) {
+    watch_fd_ = inotify_init();
+    if (watch_fd_ == -1) {
+        return EventMonitorErrorTemplates::kSystemError.Generate("cannot initialize inotify");
+    }
+    for (auto& folder : monitored_folders) {
+        auto err = AddFolderToWatch(folder, true);
+        if (err) {
+            return EventMonitorErrorTemplates::kSystemError.Generate("cannot initialize inotify: " + err->Explain());
+        }
+    }
     return nullptr;
 }
 
+Error SystemFolderWatch::ProcessInotifyEvent(struct inotify_event* i, FileEvents* events) {
+
+    printf("    wd =%2d; ", i->wd);
+    if (i->cookie > 0)
+        printf("cookie =%4d; ", i->cookie);
+
+    printf("mask = ");
+    if (i->mask & IN_ACCESS)        printf("IN_ACCESS ");
+    if (i->mask & IN_ATTRIB)        printf("IN_ATTRIB ");
+    if (i->mask & IN_CLOSE_NOWRITE) printf("IN_CLOSE_NOWRITE ");
+    if (i->mask & IN_CLOSE_WRITE)   printf("IN_CLOSE_WRITE ");
+    if (i->mask & IN_CREATE)        printf("IN_CREATE ");
+    if (i->mask & IN_DELETE)        printf("IN_DELETE ");
+    if (i->mask & IN_DELETE_SELF)   printf("IN_DELETE_SELF ");
+    if (i->mask & IN_IGNORED)       printf("IN_IGNORED ");
+    if (i->mask & IN_ISDIR)         printf("IN_ISDIR ");
+    if (i->mask & IN_MODIFY)        printf("IN_MODIFY ");
+    if (i->mask & IN_MOVE_SELF)     printf("IN_MOVE_SELF ");
+    if (i->mask & IN_MOVED_FROM)    printf("IN_MOVED_FROM ");
+    if (i->mask & IN_MOVED_TO)      printf("IN_MOVED_TO ");
+    if (i->mask & IN_OPEN)          printf("IN_OPEN ");
+    if (i->mask & IN_Q_OVERFLOW)    printf("IN_Q_OVERFLOW ");
+    if (i->mask & IN_UNMOUNT)       printf("IN_UNMOUNT ");
+    printf("\n");
+
+    if (i->len > 0)
+        printf("        name = %s\n", i->name);
+
+
+    if ((i->mask & IN_ISDIR) && ((i->mask & IN_CREATE) || (i->mask & IN_MOVED_TO))) {
+        auto it = watched_folders_paths_.find(i->wd);
+        if (it == watched_folders_paths_.end()) {
+            return EventMonitorErrorTemplates::kSystemError.Generate("cannot find monitored folder to create in " + std::to_string(
+                        i->wd));
+        }
+
+        std::string newpath = it->second + "/" + i->name;
+        auto err = AddFolderToWatch(newpath, true);
+        if (err) {
+            return err;
+        }
+    }
+
+    if ((i->mask & IN_DELETE_SELF) || ((i->mask & IN_ISDIR) && ((i->mask & IN_MOVED_FROM)))) {
+        auto it = watched_folders_paths_.find(i->wd);
+        if (it == watched_folders_paths_.end()) {
+            return EventMonitorErrorTemplates::kSystemError.Generate("cannot find monitored folder to delete " + std::to_string(
+                        i->wd));
+        }
+        std::string oldpath = it->second;
+        if (i->mask & IN_MOVED_FROM) {
+            oldpath += std::string("/") + i->name;
+            for (auto val = watched_folders_paths_.begin(); val != watched_folders_paths_.end();) {
+                if ((oldpath.size() <= val->second.size()) && std::equal(oldpath.begin(), oldpath.end(), val->second.begin())) {
+                    inotify_rm_watch(val->first, watch_fd_);
+                    GetDefaultEventMonLogger()->Debug("removed folder from monitor: " + val->second);
+                    val = watched_folders_paths_.erase(val);
+                } else {
+                    ++val;
+                }
+
+            }
+
+        } else {
+            inotify_rm_watch(it->first, watch_fd_);
+            watched_folders_paths_.erase(it);
+            GetDefaultEventMonLogger()->Debug("removed folder from monitor: " + oldpath);
+        }
+    }
+    if (!(i->mask & IN_ISDIR)) {
+        if ((i->mask & IN_CLOSE_WRITE) || (i->mask & IN_MOVED_TO)) {
+            auto it = watched_folders_paths_.find(i->wd);
+            if (it == watched_folders_paths_.end()) {
+                return EventMonitorErrorTemplates::kSystemError.Generate("cannot find monitored folder for file " + std::to_string(
+                            i->wd));
+            }
+            std::string fname = it->second + "/" + i->name;
+            FileEvent event;
+            event.type = (i->mask & IN_CLOSE_WRITE) ? EventType::closed : EventType::renamed_to;
+            event.name = fname;
+            events->emplace_back(std::move(event));
+            GetDefaultEventMonLogger()->Debug((i->mask & IN_CLOSE_WRITE) ? "file closed: " : "file moved: " + fname);
+        }
+    }
+
+
+
+    return nullptr;
+}
+
+
 FileEvents SystemFolderWatch::GetFileEventList(Error* err) {
     FileEvents events;
-    *err = nullptr;
+
+    char buffer[kBufLen]  __attribute__ ((aligned(8)));
+
+    int numRead = read(watch_fd_, buffer, sizeof(buffer));
+    if (numRead == 0) {
+        *err = TextError("readfrom inotify fd returned 0!");
+        printf("mask = ");
+
+        return events;
+    }
+
+    if (numRead == -1) {
+        *err = TextError("read from inotify fd returned -1!");
+        return events;
+    }
+
+    int nerrors = 0;
+    for (char* p = buffer; p < buffer + numRead; ) {
+        struct inotify_event* event = (struct inotify_event*) p;
+        *err = ProcessInotifyEvent(event, &events);
+        if (*err) {
+            GetDefaultEventMonLogger()->Error("error processing inotify event: " + (*err)->Explain());
+            nerrors++;
+        }
+        p += sizeof(struct inotify_event) + event->len;
+    }
+
+    if (nerrors == 0) {
+        *err = nullptr;
+    } else {
+        *err = TextError("There were " + std::to_string(nerrors) + " error(s) while processing event");
+    }
+
     return events;
 }
 
-
 }
\ No newline at end of file

producer/event_monitor_producer/src/system_folder_watch_linux.h

@@ -3,11 +3,16 @@
 
 #include <vector>
 #include <string>
+#include <map>
 
 #include "common/error.h"
 #include "preprocessor/definitions.h"
 #include "asapo_producer.h"
 #include "common.h"
+#include "io/io.h"
+#include "io/io_factory.h"
+#include <sys/inotify.h>
+#include <unistd.h>
 
 namespace asapo {
 
@@ -16,8 +21,17 @@ class SystemFolderWatch {
     VIRTUAL Error StartFolderMonitor(const std::vector<std::string>& monitored_folders);
     VIRTUAL FileEvents GetFileEventList(Error* err);
   private:
+    Error AddFolderToWatch(std::string folder, bool recursive);
+    std::unique_ptr<IO> io_{GenerateDefaultIO()};
+    Error ProcessInotifyEvent(struct inotify_event* i, FileEvents* events);
+  private:
+    static const uint64_t kBufLen  = 2000 * (sizeof(struct inotify_event) + FILENAME_MAX + 1);
+    std::map<int, std::string> watched_folders_paths_;
+    int watch_fd_ = -1;
 };
 
 }
 
+
+
 #endif //ASAPO_SYSTEM_FOLDER_WATCH_LINUX_H

tests/automatic/producer/file_monitor_producer/check_linux.sh

@@ -7,6 +7,7 @@ trap Cleanup EXIT
 Cleanup() {
 	echo cleanup
 	rm -rf test_in test_out #output
+	kill -9 $producer_id &>/dev/null
 }
 
 mkdir -p test_in test_out

tests/automatic/system_io/CMakeLists.txt

 CMAKE_MINIMUM_REQUIRED(VERSION 3.7) # needed for fixtures
 
 add_subdirectory(read_folder_content)
+add_subdirectory(read_subdirectories)
 add_subdirectory(read_file_content)
 add_subdirectory(read_string_from_file)
 add_subdirectory(ip_tcp_network)

tests/automatic/system_io/read_subdirectories/CMakeLists.txt

+set(TARGET_NAME read_subdirectories)
+set(SOURCE_FILES read_subdirectories.cpp)
+
+
+################################
+# Executable and link
+################################
+add_executable(${TARGET_NAME} ${SOURCE_FILES}  $<TARGET_OBJECTS:system_io>)
+target_link_libraries(${TARGET_NAME} test_common)
+
+#Add all necessary common libraries
+GET_PROPERTY(ASAPO_COMMON_IO_LIBRARIES GLOBAL PROPERTY ASAPO_COMMON_IO_LIBRARIES)
+target_link_libraries(${TARGET_NAME} ${ASAPO_COMMON_IO_LIBRARIES})
+
+target_include_directories(${TARGET_NAME} PUBLIC ${ASAPO_CXX_COMMON_INCLUDE_DIR})
+set_target_properties(${TARGET_NAME} PROPERTIES LINKER_LANGUAGE CXX)
+
+################################
+# Testing
+################################
+
+add_test_setup_cleanup(${TARGET_NAME})
+IF(WIN32)
+    add_integration_test(${TARGET_NAME} list_folders "test test\\subtest3test\\subtest3\\subtest4test\\subtest1test\\subtest1\\subtest2")
+ELSE()
+    add_integration_test(${TARGET_NAME} list_folders "test test/subtest3test/subtest3/subtest4test/subtest1test/subtest1/subtest2")
+ENDIF(WIN32)
+
+
+add_integration_test(${TARGET_NAME} foldernotfound "test_notexist Nosuchfileordirectory:test_notexist")
+add_integration_test(${TARGET_NAME} foldernoaccess "test_noaccess1 Permissiondenied:test_noaccess1")
+

tests/automatic/system_io/read_subdirectories/cleanup_linux.sh

+#!/usr/bin/env bash
+
+rm -rf test
+rmdir test_noaccess1

tests/automatic/system_io/read_subdirectories/cleanup_windows.bat

+#rmdir /S /Q test
+icacls test_noaccess1 /grant:r users:D
+rmdir /S /Q test_noaccess1

tests/automatic/system_io/read_subdirectories/read_subdirectories.cpp

+#include <iostream>
+
+#include "io/io_factory.h"
+#include "testing.h"
+
+using asapo::IO;
+using asapo::Error;
+
+
+using asapo::M_AssertEq;
+using asapo::M_AssertContains;
+
+int main(int argc, char* argv[]) {
+    if (argc != 3) {
+        std::cout << "Wrong number of arguments" << std::endl;
+        return 1;
+    }
+    std::string expect{argv[2]};
+
+    Error err;
+    auto io = std::unique_ptr<IO> {asapo::GenerateDefaultIO() };
+    auto subdirs = io->GetSubDirectories(argv[1], &err);
+
+    std::string result{};
+    if (err == nullptr) {
+        for(auto folder : subdirs) {
+            result += folder;
+        }
+    } else {
+        result = err->Explain();
+    }
+
+    M_AssertContains(result, expect);
+
+    return 0;
+}

tests/automatic/system_io/read_subdirectories/setup_linux.sh

+#!/usr/bin/env bash
+
+mkdir -p test/subtest1/subtest2
+sleep 0.1
+mkdir -p test/subtest3/subtest4/
+sleep 0.1
+
+mkdir test_noaccess1
+chmod -rx test_noaccess1
+

tests/automatic/system_io/read_subdirectories/setup_windows.bat

+mkdir test
+mkdir test\subtest1
+mkdir test\subtest1\subtest2
+
+mkdir test\subtest3
+mkdir test\subtest3\subtest4
+
+ping 1.0.0.0 -n 1 -w 100 > nul
+
+mkdir test_noaccess1
+icacls test_noaccess1 /deny users:D
+
+
+