Skip to content

Feature/refactor event processor #12

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

Merged
d2phap merged 2 commits into from
Jan 5, 2023
Merged

Feature/refactor event processor #12

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6fd5dd3
extract spam warning
samba2 Dec 20, 2022
f2c3e69
small refactoring
samba2 Dec 20, 2022
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
91 changes: 46 additions & 45 deletions Source/FileWatcherEx/Helpers/EventProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,9 @@ internal class EventProcessor
private const int EventDelay = 50;

/// <summary>
/// Warn after certain time span of event spam (in ticks)
/// Warn after certain time span of event spam
/// </summary>
private const int EventSpamWarningThreshold = 60 * 1000 * 10000;
private readonly TimeSpan _eventSpamWarningThreshold = TimeSpan.FromMinutes(1);

private readonly object _lock = new();
private Task? _delayTask = null;
Expand Down Expand Up @@ -42,20 +42,7 @@ public void ProcessEvent(FileChangedEvent fileEvent)
lock (_lock)
{
var now = DateTime.Now.Ticks;

// Check for spam
if (_events.Count == 0)
{
_spamWarningLogged = false;
_spamCheckStartTime = now;
}
else if (!_spamWarningLogged && _spamCheckStartTime + EventSpamWarningThreshold < now)
{
_spamWarningLogged = true;
_logger(string.Format(
"Warning: Watcher is busy catching up with {0} file changes in 60 seconds. Latest path is '{1}'",
_events.Count, fileEvent.FullPath));
}
WarnForSpam(fileEvent, now);

// Add into our queue
_events.Add(fileEvent);
Expand All @@ -64,40 +51,54 @@ public void ProcessEvent(FileChangedEvent fileEvent)
// Process queue after delay
if (_delayTask == null)
{
// Create function to buffer events
void Func(Task value)
// Start function after delay
_delayStarted = _lastEventTime;
_delayTask = Task.Delay(EventDelay).ContinueWith(HandleEventsFunc);
}
}
}

private void HandleEventsFunc(Task _)
{
lock (_lock)
{
// Check if another event has been received in the meantime
if (_delayStarted == _lastEventTime)
{
// Normalize and handle
var normalized = new EventNormalizer().Normalize(_events.ToArray());
foreach (var ev in normalized)
{
lock (_lock)
{
// Check if another event has been received in the meantime
if (_delayStarted == _lastEventTime)
{
// Normalize and handle
var normalized = new EventNormalizer().Normalize(_events.ToArray());
foreach (var e in normalized)
{
_handleEvent(e);
}

// Reset
_events.Clear();
_delayTask = null;
}

// Otherwise we have received a new event while this task was
// delayed and we reschedule it.
else
{
_delayStarted = _lastEventTime;
_delayTask = Task.Delay(EventDelay).ContinueWith(Func);
}
}
_handleEvent(ev);
}

// Start function after delay
// Reset
_events.Clear();
_delayTask = null;
}

// Otherwise we have received a new event while this task was
// delayed and we reschedule it.
else
{
_delayStarted = _lastEventTime;
_delayTask = Task.Delay(EventDelay).ContinueWith(Func);
_delayTask = Task.Delay(EventDelay).ContinueWith(HandleEventsFunc);
}
}
}

private void WarnForSpam(FileChangedEvent fileEvent, long now)
{
if (_events.Count == 0)
{
_spamWarningLogged = false;
_spamCheckStartTime = now;
}
else if (! _spamWarningLogged && _spamCheckStartTime + _eventSpamWarningThreshold.Ticks < now)
{
_spamWarningLogged = true;
_logger($"Warning: Watcher is busy catching up with {_events.Count} file changes " +
$"in {_eventSpamWarningThreshold.TotalSeconds} seconds. Latest path is '{fileEvent.FullPath}'");
}
}
}