continue windows folder monitor

producer/event_monitor_producer/CMakeLists.txt

@@ -10,6 +10,8 @@ IF(WIN32)
     set(SOURCE_FILES ${SOURCE_FILES} src/system_folder_watch_windows.cpp
             src/single_folder_watch_windows.cpp
             src/watch_io.cpp
+            src/shared_event_list.cpp
+            src/win_event.cpp
             )
 ELSEIF(UNIX)
     set(SOURCE_FILES ${SOURCE_FILES} src/system_folder_watch_linux.cpp src/inotify_event.cpp src/inotify_linux.cpp)

producer/event_monitor_producer/src/inotify_event.h

@@ -26,7 +26,7 @@ class InotifyEvent {
     void Print() const;
   private:
     const struct inotify_event* inotify_event_;
-    const std::map<int, std::string> watched_folders_paths_;
+    const std::map<int, std::string>& watched_folders_paths_;
 };
 
 }

producer/event_monitor_producer/src/shared_event_list.cpp

+#include "shared_event_list.h"
+
+namespace asapo {
+
+FilesToSend SharedEventList::GetAndClearEvents() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    FilesToSend events = std::move(events_);
+    events_.clear();
+    return events;
+}
+void SharedEventList::AddEvent(std::string event) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    events_.emplace_back(std::move(event));
+}
+}

producer/event_monitor_producer/src/shared_event_list.h

+#ifndef ASAPO_SHARED_EVENT_LIST_H
+#define ASAPO_SHARED_EVENT_LIST_H
+
+#include <string>
+#include <vector>
+#include <mutex>
+
+#include "common.h"
+
+namespace asapo {
+
+class SharedEventList {
+ public:
+  FilesToSend GetAndClearEvents();
+  void AddEvent(std::string event);
+ private:
+  std::mutex mutex_;
+  FilesToSend events_;
+};
+
+}
+
+#endif //ASAPO_SHARED_EVENT_LIST_H

producer/event_monitor_producer/src/single_folder_watch_windows.cpp

 #include "single_folder_watch_windows.h"
 
 #include "eventmon_logger.h"
+
+#include <iostream>
+#include <string>
+
 namespace asapo {
 
-SingleFolderWatch::SingleFolderWatch(std::string root_folder, std::string folder) : watch_io__{new WatchIO()},
+SingleFolderWatch::SingleFolderWatch(std::string root_folder, std::string folder,SharedEventList* event_list) :
+                                                                                    watch_io__{new WatchIO()},
                                                                                     log__{GetDefaultEventMonLogger()},
                                                                                     root_folder_{std::move(root_folder)},
-                                                                                        folder_{std::move(folder)}
+                                                                                        folder_{std::move(folder)},
+                                                                                         buffer_{new char[kBufLen]},
+                                                                                           event_list_{event_list}
+
                                                                                     {
 }
 
@@ -22,9 +30,32 @@ Error SingleFolderWatch::Init()  {
 }
 
 void SingleFolderWatch::Watch() {
-    if (!Init()) {
+    if (Init()!=nullptr) {
         return;
     }
+
+    DWORD bytes_read = 0;
+    auto err =watch_io__->ReadDirectoryChanges(handle_,buffer_.get(),kBufLen,&bytes_read);
+    if (err == nullptr) {
+        ProcessEvents(bytes_read);
+    }
+}
+Error SingleFolderWatch::ProcessEvent(const WinEvent &event) {
+    event.Print();
+    event_list_->AddEvent(event.FileName());
+    return nullptr;
+}
+void SingleFolderWatch::ProcessEvents(DWORD bytes_to_read) {
+    for (char* p = buffer_.get(); p < buffer_.get() + bytes_to_read; ) {
+        WinEvent event{(FILE_NOTIFY_INFORMATION*) p};
+        ProcessEvent(event);
+        p += event.Offset();
+        if (event.Offset() == 0) {
+            break;
+        }
+    }
 }
 
 }
+
+

producer/event_monitor_producer/src/single_folder_watch_windows.h

@@ -2,15 +2,23 @@
 #define ASAPO_SINGLE_FOLDER_MONITOR_H
 
 #include <string>
+#include <windows.h>
+#include <winnt.h>
+
 
 #include "watch_io.h"
 #include "logger/logger.h"
+#include "shared_event_list.h"
+#include "win_event.h"
 
 namespace asapo {
 
+const uint64_t kBufLen  = 1000 * (sizeof(FILE_NOTIFY_INFORMATION) + FILENAME_MAX + 1);
+
+
 class SingleFolderWatch {
   public:
-    explicit SingleFolderWatch(std::string root_folder,std::string folder);
+    explicit SingleFolderWatch(std::string root_folder,std::string folder,SharedEventList* event_list);
     void Watch();
     std::unique_ptr<WatchIO> watch_io__;
     const AbstractLogger* log__;
@@ -19,6 +27,10 @@ class SingleFolderWatch {
   std::string folder_;
   Error Init();
   HANDLE handle_;
+  SharedEventList* event_list_;
+  std::unique_ptr<char[]> buffer_;
+  Error ProcessEvent(const WinEvent& event);
+  void ProcessEvents(DWORD bytes_to_read);
 };
 
 }

producer/event_monitor_producer/src/system_folder_watch_windows.cpp

@@ -11,14 +11,15 @@ namespace asapo {
 Error SystemFolderWatch::StartFolderMonitor(const std::string& root_folder,
                                             const std::vector<std::string>& monitored_folders) {
     for (auto& folder:monitored_folders ) {
-    auto thread = io__->NewThread([root_folder, folder] {
-      auto folder_watch = std::unique_ptr<SingleFolderWatch>(new SingleFolderWatch(root_folder, folder));
+    auto thread = io__->NewThread([root_folder, folder,this] {
+      auto folder_watch = std::unique_ptr<SingleFolderWatch>(new SingleFolderWatch(root_folder, folder,&event_list_));
       folder_watch->Watch();
     });
 
-    if (thread) {
-        thread->detach();
-    }
+    threads_.emplace_back(std::move(thread));
+//    if (thread) {
+//        thread->detach();
+//    }
     }
 
     return nullptr;

producer/event_monitor_producer/src/system_folder_watch_windows.h

@@ -9,10 +9,12 @@
 #include "asapo_producer.h"
 #include "common.h"
 #include "io/io.h"
-
+#include "shared_event_list.h"
 
 namespace asapo {
 
+
+
 class SystemFolderWatch {
   public:
   SystemFolderWatch();
@@ -21,7 +23,8 @@ class SystemFolderWatch {
   VIRTUAL FilesToSend GetFileList(Error* err);
   std::unique_ptr<IO> io__;
   private:
-
+    SharedEventList event_list_;
+    std::vector<std::unique_ptr<std::thread>> threads_;
 };
 
 }

producer/event_monitor_producer/src/watch_io.cpp

@@ -20,5 +20,18 @@ HANDLE WatchIO::Init(const char* folder, Error* err) {
 WatchIO::WatchIO() :io_{GenerateDefaultIO()}{
 
 }
+Error WatchIO::ReadDirectoryChanges(HANDLE handle, LPVOID buffer, DWORD buffer_length, LPDWORD bytes_returned) {
+    DWORD filter = FILE_NOTIFY_CHANGE_FILE_NAME |
+        FILE_NOTIFY_CHANGE_LAST_WRITE;
+    auto res = ReadDirectoryChangesW(handle,buffer,buffer_length,true,filter,bytes_returned,nullptr,nullptr);
+    printf("after read changes\n");
+    if (res) {
+        printf("after read changes ok\n");
+        return nullptr;
+    } else {
+        printf("after read problem\n");
+        return io_->GetLastError();
+    }
+}
 
 }
\ No newline at end of file

producer/event_monitor_producer/src/watch_io.h

@@ -13,6 +13,7 @@ class WatchIO {
  public:
   explicit WatchIO();
   VIRTUAL HANDLE Init(const char* folder, Error* err);
+  VIRTUAL Error ReadDirectoryChanges(HANDLE handle,LPVOID buffer, DWORD buffer_length,LPDWORD bytes_returned);
  private:
   std::unique_ptr<IO>io_;
 };

producer/event_monitor_producer/src/win_event.cpp

+#include "win_event.h"
+
+#include <vector>
+
+
+namespace asapo {
+
+WinEvent::WinEvent(const FILE_NOTIFY_INFORMATION* win_event):win_event_{win_event} {
+
+}
+std::string WinEvent::FileName() const {
+    std::size_t len = win_event_->FileNameLength/ sizeof(WCHAR);
+    std::vector<char> buffer(len + 1);
+    buffer[len]=0;
+//    std::locale loc("");
+//    std::use_facet<std::ctype<wchar_t> >(loc).narrow(win_event_->FileName, win_event_->FileName + len, '_', &buffer[0]);
+    for (size_t i=0;i<len;i++) {
+        buffer[i] = (char)win_event_->FileName[i];
+    }
+    return std::string(&buffer[0], &buffer[len]);
+}
+
+size_t WinEvent::Offset()const {
+    return win_event_->NextEntryOffset;
+}
+
+void WinEvent::Print() const{
+    printf("\nNew Event: ");
+    if (win_event_->Action == FILE_ACTION_ADDED) printf("FILE_ACTION_ADDED ");
+    if (win_event_->Action == FILE_ACTION_REMOVED) printf("FILE_ACTION_REMOVED ");
+    if (win_event_->Action == FILE_ACTION_MODIFIED) printf("FILE_ACTION_MODIFIED ");
+    if (win_event_->Action == FILE_ACTION_RENAMED_OLD_NAME) printf("FILE_ACTION_RENAMED_OLD_NAME ");
+    if (win_event_->Action == FILE_ACTION_RENAMED_NEW_NAME) printf("FILE_ACTION_RENAMED_NEW_NAME ");
+    printf("\n");
+    if (win_event_->FileNameLength > 0)
+        printf("Filename: %s\n",FileName().c_str());
+
+}
+}

producer/event_monitor_producer/src/win_event.h

+#ifndef ASAPO_WIN_EVENT_H
+#define ASAPO_WIN_EVENT_H
+
+#include <windows.h>
+#include <winnt.h>
+#include <string>
+
+namespace asapo {
+
+class WinEvent {
+ public:
+  WinEvent(const FILE_NOTIFY_INFORMATION* win_event);
+  size_t Offset() const;
+  void Print() const;
+  std::string FileName() const ;
+ private:
+  const FILE_NOTIFY_INFORMATION* win_event_;
+
+};
+
+}
+
+#endif //ASAPO_WIN_EVENT_H

producer/event_monitor_producer/unittests/mock_watch_io.h

@@ -18,6 +18,13 @@ class MockWatchIO : public WatchIO {
    }
 
   MOCK_METHOD2(Init_t, HANDLE (const char* folder, ErrorInterface** err));
+
+  Error ReadDirectoryChanges(HANDLE handle,LPVOID buffer, DWORD buffer_length,LPDWORD bytes_returned) override {
+      return Error{ReadDirectoryChanges_t(handle,buffer,buffer_length,bytes_returned)};
+  }
+
+  MOCK_METHOD4(ReadDirectoryChanges_t, ErrorInterface* (HANDLE handle,LPVOID buffer, DWORD buffer_length,LPDWORD bytes_returned));
+
 };
 
 }

producer/event_monitor_producer/unittests/test_single_folder_watch_windows.cpp

 #include <gtest/gtest.h>
 #include <gmock/gmock.h>
+#include <chrono>
+
 
 #include "../src/single_folder_watch_windows.h"
 #include "../src/event_monitor_error.h"
@@ -36,7 +38,7 @@ namespace {
 
 
 TEST(SingleFolderWatch, Constructor) {
-    SingleFolderWatch watch{"",""};
+    SingleFolderWatch watch{"","",nullptr};
     ASSERT_THAT(dynamic_cast<asapo::WatchIO*>(watch.watch_io__.get()), Ne(nullptr));
 }
 
@@ -60,18 +62,58 @@ class SingleFolderWatchTests : public testing::Test {
     std::string expected_root_folder = "c:\\tmp";
     std::string expected_folder{"test1"};
     HANDLE expected_handle = HANDLE(1);
-    SingleFolderWatch watch{expected_root_folder,expected_folder};
+    asapo::SharedEventList event_list;
+    SingleFolderWatch watch{expected_root_folder,expected_folder,&event_list};
+    char* buffer;
+    DWORD cur_buffer_pointer = 0;
     void SetUp() override {
         watch.watch_io__ = std::unique_ptr<asapo::WatchIO> {&mock_watch_io};
         watch.log__ = &mock_logger;
+        buffer = new (char[asapo::kBufLen]);
     }
     void TearDown() override {
         watch.watch_io__.release();
+        delete[] buffer;
+    }
+    void ExpectInit();
+    void ExpectRead();
+    DWORD AddEventToBuffer(std::string filename, DWORD action);
+
+  };
+
+DWORD SingleFolderWatchTests::AddEventToBuffer(std::string filename, DWORD action) {
+    size_t filename_size = filename.size();
+    DWORD size = sizeof(FILE_NOTIFY_INFORMATION) + filename_size*sizeof(WCHAR);
+    char* buf = (char*) malloc(size);
+    FILE_NOTIFY_INFORMATION* event = (FILE_NOTIFY_INFORMATION*) buf;
+    event->NextEntryOffset = size;
+    event->Action = action;
+    for (size_t i=0;i<filename_size;i++) {
+        event->FileName[i] = filename[i];
     }
-};
+    event->FileNameLength = filename_size* sizeof(WCHAR);
+    memcpy(buffer + cur_buffer_pointer, event, size);
+    cur_buffer_pointer += size;
+    free(buf);
+    return size;
+}
 
+ACTION_P(A_CopyBuf, buffer) {
+    memcpy(arg1, buffer, asapo::kBufLen);
+}
 
-TEST_F(SingleFolderWatchTests, InitWatchOnWatch) {
+
+void SingleFolderWatchTests::ExpectRead() {
+    EXPECT_CALL(mock_watch_io, ReadDirectoryChanges_t(expected_handle, _, asapo::kBufLen,_))
+        .WillOnce(DoAll(
+            A_CopyBuf(buffer),
+            SetArgPointee<3>(cur_buffer_pointer),
+            Return(nullptr))
+        );
+}
+
+
+void SingleFolderWatchTests::ExpectInit() {
     EXPECT_CALL(mock_watch_io, Init_t(StrEq(expected_root_folder+asapo::kPathSeparator+expected_folder),_)).
         WillOnce(DoAll(
         SetArgPointee<1>(nullptr),
@@ -79,6 +121,11 @@ TEST_F(SingleFolderWatchTests, InitWatchOnWatch) {
                  )
     );
 
+
+}
+
+TEST_F(SingleFolderWatchTests, InitWatchOnWatch) {
+    ExpectInit();
     watch.Watch();
 }
 
@@ -101,6 +148,23 @@ TEST_F(SingleFolderWatchTests, InitErrorOnWatch) {
     watch.Watch();
 }
 
+TEST_F(SingleFolderWatchTests, WatchReadsDirectoryEvents) {
+    ExpectInit();
+    AddEventToBuffer("test",FILE_ACTION_ADDED);
+    AddEventToBuffer("test2",FILE_ACTION_MODIFIED);
+
+    ExpectRead();
+    watch.Watch();
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(30));
+    auto files = event_list.GetAndClearEvents();
+
+    ASSERT_THAT(files.size(), Eq(2));
+    ASSERT_THAT(files[0], StrEq("test"));
+    ASSERT_THAT(files[1], StrEq("test2"));
+
+
+}
 
 
 

producer/event_monitor_producer/unittests/test_system_folder_watch_windows.cpp

@@ -43,7 +43,7 @@ FileInfos CreateTestFileInfos() {
     fi.size = 100;
     fi.name = "file1";
     file_infos.push_back(fi);
-    fi.name = "subfolder/file2";
+    fi.name = "subfolder\\file2";
     file_infos.push_back(fi);
     return file_infos;
 }
@@ -54,9 +54,9 @@ class SystemFolderWatchTests : public testing::Test {
     Error err;
     ::testing::NiceMock<asapo::MockIO> mock_io;
     SystemFolderWatch watch{};
-    std::string expected_root_folder = "/tmp";
+    std::string expected_root_folder = "c:\\tmp";
     std::vector<std::string> expected_folders{"test1", "test2"};
-    void SetUp() override {
+  void SetUp() override {
         watch.io__ = std::unique_ptr<asapo::IO> {&mock_io};
     }
     void TearDown() override {
@@ -64,7 +64,7 @@ class SystemFolderWatchTests : public testing::Test {
     }
 };
 
-TEST_F(SystemFolderWatchTests, ErrorInitInotifyStartMonitoring) {
+TEST_F(SystemFolderWatchTests,StartMonitoring) {
 
 
     EXPECT_CALL(mock_io, NewThread_t(_)).Times(expected_folders.size()).