Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Make event queue shared between nsfw and native implementation #77

Merged
merged 2 commits into from
Jun 21, 2019
Merged

Make event queue shared between nsfw and native implementation #77

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
558427a
Make event queue shared between nsfw and native implementation
implausible Jun 21, 2019
83c97b4
Add timeout between mkdir steps to increase test stability
implausible Jun 21, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 6 additions & 9 deletions includes/NSFW.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#ifndef NSFW_H
#define NSFW_H

#include "Queue.h"
#include "NativeInterface.h"
#include <nan.h>
#include <uv.h>
Expand All @@ -13,12 +14,15 @@ class NSFW : public ObjectWrap {
public:
static NAN_MODULE_INIT(Init);

static void cleanupEventCallback(void *arg);
static void fireErrorCallback(uv_async_t *handle);
static void fireEventCallback(uv_async_t *handle);
static void pollForEvents(void *arg);

Persistent<v8::Object> mPersistentHandle;
private:
NSFW(uint32_t debounceMS, std::string path, Callback *eventCallback, Callback *errorCallback);
~NSFW();

uint32_t mDebounceMS;
uv_async_t mErrorCallbackAsync;
uv_async_t mEventCallbackAsync;
Expand All @@ -30,20 +34,13 @@ class NSFW : public ObjectWrap {
std::string mPath;
uv_thread_t mPollThread;
std::atomic<bool> mRunning;
private:
NSFW(uint32_t debounceMS, std::string path, Callback *eventCallback, Callback *errorCallback);
~NSFW();
std::shared_ptr<EventQueue> mQueue;

struct ErrorBaton {
NSFW *nsfw;
std::string error;
};

struct EventBaton {
NSFW *nsfw;
std::unique_ptr<std::vector<std::unique_ptr<Event>>> events;
};

static NAN_METHOD(JSNew);

static NAN_METHOD(Start);
Expand Down
4 changes: 1 addition & 3 deletions includes/NativeInterface.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,14 @@ using NativeImplementation = InotifyService;

class NativeInterface {
public:
NativeInterface(const std::string &path);
NativeInterface(const std::string &path, std::shared_ptr<EventQueue> queue);
~NativeInterface();

std::string getError();
std::unique_ptr<std::vector<std::unique_ptr<Event>>> getEvents();
bool hasErrored();
bool isWatching();

private:
std::shared_ptr<EventQueue> mQueue;
std::unique_ptr<NativeImplementation> mNativeInterface;
};

Expand Down
7 changes: 6 additions & 1 deletion js/spec/index-spec.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,15 @@ const nsfw = require('../src/');
const path = require('path');
const fse = require('fs-extra');
const exec = require('executive');
const { promisify } = require('util');

jasmine.DEFAULT_TIMEOUT_INTERVAL = 120000;

const DEBOUNCE = 1000;
const TIMEOUT_PER_STEP = 3000;

const timeout = promisify(setTimeout);

describe('Node Sentinel File Watcher', function() {
const workDir = path.resolve('./mockfs');

Expand Down Expand Up @@ -396,7 +399,9 @@ describe('Node Sentinel File Watcher', function() {
return paths.reduce((chain, dir) => {
directory = path.join(directory, dir);
const nextDirectory = directory;
return chain.then(() => fse.mkdir(nextDirectory));
return chain
.then(() => fse.mkdir(nextDirectory))
.then(() => timeout(60));
}, Promise.resolve());
})
.then(() => fse.open(path.join(directory, file), 'w'))
Expand Down
62 changes: 22 additions & 40 deletions src/NSFW.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@ NSFW::NSFW(uint32_t debounceMS, std::string path, Callback *eventCallback, Callb
mInterface(NULL),
mInterfaceLockValid(false),
mPath(path),
mRunning(false) {
mRunning(false),
mQueue(std::make_shared<EventQueue>())
{
HandleScope scope;
v8::Local<v8::Object> obj = New<v8::Object>();
mPersistentHandle.Reset(obj);
Expand All @@ -37,11 +39,6 @@ NSFW::~NSFW() {
}
}

void NSFW::cleanupEventCallback(void *arg) {
EventBaton *baton = (EventBaton *)arg;
delete baton;
}

void NSFW::fireErrorCallback(uv_async_t *handle) {
Nan::HandleScope scope;
ErrorBaton *baton = (ErrorBaton *)handle->data;
Expand All @@ -54,33 +51,27 @@ void NSFW::fireErrorCallback(uv_async_t *handle) {

void NSFW::fireEventCallback(uv_async_t *handle) {
Nan::HandleScope scope;
EventBaton *baton = (EventBaton *)handle->data;
if (baton->events->empty()) {
uv_thread_t cleanup;
uv_thread_create(&cleanup, NSFW::cleanupEventCallback, baton);

#if defined(__APPLE_CC__) || defined(__linux__) || defined(__FreeBSD__)
pthread_detach(cleanup);
#endif

NSFW *nsfw = (NSFW *)handle->data;
auto events = nsfw->mQueue->dequeueAll();
if (events == nullptr) {
return;
}

v8::Local<v8::Array> eventArray = New<v8::Array>((int)baton->events->size());
v8::Local<v8::Array> eventArray = New<v8::Array>((int)events->size());

for (unsigned int i = 0; i < baton->events->size(); ++i) {
for (unsigned int i = 0; i < events->size(); ++i) {
v8::Local<v8::Object> jsEvent = New<v8::Object>();


jsEvent->Set(New<v8::String>("action").ToLocalChecked(), New<v8::Number>((*baton->events)[i]->type));
jsEvent->Set(New<v8::String>("directory").ToLocalChecked(), New<v8::String>((*baton->events)[i]->fromDirectory).ToLocalChecked());
jsEvent->Set(New<v8::String>("action").ToLocalChecked(), New<v8::Number>((*events)[i]->type));
jsEvent->Set(New<v8::String>("directory").ToLocalChecked(), New<v8::String>((*events)[i]->fromDirectory).ToLocalChecked());

if ((*baton->events)[i]->type == RENAMED) {
jsEvent->Set(New<v8::String>("oldFile").ToLocalChecked(), New<v8::String>((*baton->events)[i]->fromFile).ToLocalChecked());
jsEvent->Set(New<v8::String>("newDirectory").ToLocalChecked(), New<v8::String>((*baton->events)[i]->toDirectory).ToLocalChecked());
jsEvent->Set(New<v8::String>("newFile").ToLocalChecked(), New<v8::String>((*baton->events)[i]->toFile).ToLocalChecked());
if ((*events)[i]->type == RENAMED) {
jsEvent->Set(New<v8::String>("oldFile").ToLocalChecked(), New<v8::String>((*events)[i]->fromFile).ToLocalChecked());
jsEvent->Set(New<v8::String>("newDirectory").ToLocalChecked(), New<v8::String>((*events)[i]->toDirectory).ToLocalChecked());
jsEvent->Set(New<v8::String>("newFile").ToLocalChecked(), New<v8::String>((*events)[i]->toFile).ToLocalChecked());
} else {
jsEvent->Set(New<v8::String>("file").ToLocalChecked(), New<v8::String>((*baton->events)[i]->fromFile).ToLocalChecked());
jsEvent->Set(New<v8::String>("file").ToLocalChecked(), New<v8::String>((*events)[i]->fromFile).ToLocalChecked());
}

eventArray->Set(i, jsEvent);
Expand All @@ -90,14 +81,7 @@ void NSFW::fireEventCallback(uv_async_t *handle) {
eventArray
};

baton->nsfw->mEventCallback->Call(1, argv);

uv_thread_t cleanup;
uv_thread_create(&cleanup, NSFW::cleanupEventCallback, baton);

#if defined(__APPLE_CC__) || defined(__linux__) || defined(__FreeBSD__)
pthread_detach(cleanup);
#endif
nsfw->mEventCallback->Call(1, argv);
}

void NSFW::pollForEvents(void *arg) {
Expand All @@ -116,18 +100,14 @@ void NSFW::pollForEvents(void *arg) {
uv_mutex_unlock(&nsfw->mInterfaceLock);
break;
}
auto events = nsfw->mInterface->getEvents();
if (events == NULL) {

if (nsfw->mQueue->count() == 0) {
uv_mutex_unlock(&nsfw->mInterfaceLock);
sleep_for_ms(50);
continue;
}

EventBaton *baton = new EventBaton;
baton->nsfw = nsfw;
baton->events = std::move(events);

nsfw->mEventCallbackAsync.data = (void *)baton;
nsfw->mEventCallbackAsync.data = (void *)nsfw;
uv_async_send(&nsfw->mEventCallbackAsync);

uv_mutex_unlock(&nsfw->mInterfaceLock);
Expand Down Expand Up @@ -230,7 +210,8 @@ void NSFW::StartWorker::Execute() {
return;
}

mNSFW->mInterface = new NativeInterface(mNSFW->mPath);
mNSFW->mQueue->clear();
mNSFW->mInterface = new NativeInterface(mNSFW->mPath, mNSFW->mQueue);
if (mNSFW->mInterface->isWatching()) {
mNSFW->mRunning = true;
uv_thread_create(&mNSFW->mPollThread, NSFW::pollForEvents, mNSFW);
Expand Down Expand Up @@ -306,6 +287,7 @@ void NSFW::StopWorker::Execute() {
uv_mutex_lock(&mNSFW->mInterfaceLock);
delete mNSFW->mInterface;
mNSFW->mInterface = NULL;
mNSFW->mQueue->clear();

uv_mutex_unlock(&mNSFW->mInterfaceLock);
}
Expand Down
9 changes: 2 additions & 7 deletions src/NativeInterface.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
#include "../includes/NativeInterface.h"

NativeInterface::NativeInterface(const std::string &path) {
mQueue = std::make_shared<EventQueue>();
mNativeInterface.reset(new NativeImplementation(mQueue, path));
NativeInterface::NativeInterface(const std::string &path, std::shared_ptr<EventQueue> queue) {
mNativeInterface.reset(new NativeImplementation(queue, path));
}

NativeInterface::~NativeInterface() {
Expand All @@ -13,10 +12,6 @@ std::string NativeInterface::getError() {
return mNativeInterface->getError();
}

std::unique_ptr<std::vector<std::unique_ptr<Event>>> NativeInterface::getEvents() {
return mQueue->dequeueAll();
}

bool NativeInterface::hasErrored() {
return mNativeInterface->hasErrored();
}
Expand Down