<commit_before>﻿using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System;
using System.IO;
using Utility.Observable;
using Interfaces;
using NLog;
using System.Linq;
using System.Timers;
using System.Threading.Tasks;
using DW.ELA.LogModel;
using DW.ELA.Interfaces;

namespace Controller
{
    /// <summary>
    /// This class runs a background thread to monitor and notify consumers (observers) of new log lines
    /// </summary>
    public class JsonLogMonitor : AbstractObservable<LogEvent>, ILogRealTimeDataSource
    {
        private readonly FileSystemWatcher fileWatcher;
        private string CurrentFile;
        private readonly string LogDirectory;
        private static readonly ILogger logger = LogManager.GetCurrentClassLogger();
        private long filePosition;
        private object @lock = new object();
        private readonly Timer logFlushTimer = new Timer();

        public JsonLogMonitor(ILogDirectoryNameProvider logDirectoryProvider)
        {
            LogDirectory = logDirectoryProvider.Directory;
            fileWatcher = new FileSystemWatcher(LogDirectory);

            fileWatcher.Changed += FileWatcher_Changed;
            fileWatcher.Created += FileWatcher_Created;
            fileWatcher.NotifyFilter = NotifyFilters.CreationTime |
                                       NotifyFilters.FileName |
                                       NotifyFilters.LastWrite |
                                       NotifyFilters.Size;

            logFlushTimer.AutoReset = true;
            logFlushTimer.Interval = 5000; // sometimes the filesystem event does not trigger
            logFlushTimer.Elapsed += (o, e) => Task.Factory.StartNew(() => Update(false));
            logFlushTimer.Enabled = true;

            CurrentFile = LogEnumerator.GetLogFiles(LogDirectory)[0];
            filePosition = new FileInfo(CurrentFile).Length;
            Update(false);
            fileWatcher.EnableRaisingEvents = true;
            logger.Debug("Started monitoring on folder {0}", LogDirectory);
        }

        private void FileWatcher_Created(object sender, FileSystemEventArgs e)
        {
            Update(checkOtherFiles: true);
        }

        private void FileWatcher_Changed(object sender, FileSystemEventArgs e)
        {
            if (Path.GetExtension(e.FullPath) != ".log")
                return;
            Update(checkOtherFiles: e.FullPath != CurrentFile);
        }

        private void Update(bool checkOtherFiles)
        {
            lock (@lock)
                try
                {
                    filePosition = ReadFileFromPosition(CurrentFile, filePosition);
                    if (checkOtherFiles)
                    {
                        var filesToScan = LogEnumerator.GetLogFiles(LogDirectory)
                            .Take(10) // max 10 'missed' files to upload
                            .TakeWhile(f => f != CurrentFile) // only until we meet our file
                            .Reverse() // back to chronological order
                            .ToArray();
                        foreach (var file in filesToScan)
                            filePosition = ReadFileFromPosition(file, filePosition);
                        CurrentFile = filesToScan.Last();
                    }
                }
                catch (Exception e)
                {
                    logger.Warn(e, "Error while reading log file");
                }
        }

        private long ReadFileFromPosition(string file, long filePosition)
        {
            using (var fileReader = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
            {
                fileReader.Position = filePosition;
                using (var textReader = new StreamReader(fileReader))
                using (var jsonReader = new JsonTextReader(textReader) { SupportMultipleContent = true })
                {
                    JsonSerializer serializer = new JsonSerializer();
                    while (jsonReader.Read())
                    {
                        var record = serializer.Deserialize(jsonReader);
                        // Sometimes serializer gives out the json as string instead of JObject
                        var @object = record as JObject ?? JObject.Parse(record as string);
                        OnNext(LogEventConverter.Convert(@object));
                        logger.Debug("Received event: {0}", @object.ToString());
                    }
                    return fileReader.Position;
                }
            }
        }

        #region IDisposable Support<commit_msg>Fix: single parsing errors should not kill Json parser<commit_after>﻿using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System;
using System.IO;
using Utility.Observable;
using Interfaces;
using NLog;
using System.Linq;
using System.Timers;
using System.Threading.Tasks;
using DW.ELA.LogModel;
using DW.ELA.Interfaces;

namespace Controller
{
    /// <summary>
    /// This class runs a background thread to monitor and notify consumers (observers) of new log lines
    /// </summary>
    public class JsonLogMonitor : AbstractObservable<LogEvent>, ILogRealTimeDataSource
    {
        private readonly FileSystemWatcher fileWatcher;
        private string CurrentFile;
        private readonly string LogDirectory;
        private static readonly ILogger logger = LogManager.GetCurrentClassLogger();
        private long filePosition;
        private object @lock = new object();
        private readonly Timer logFlushTimer = new Timer();

        public JsonLogMonitor(ILogDirectoryNameProvider logDirectoryProvider)
        {
            LogDirectory = logDirectoryProvider.Directory;
            fileWatcher = new FileSystemWatcher(LogDirectory);

            fileWatcher.Changed += FileWatcher_Changed;
            fileWatcher.Created += FileWatcher_Created;
            fileWatcher.NotifyFilter = NotifyFilters.CreationTime |
                                       NotifyFilters.FileName |
                                       NotifyFilters.LastWrite |
                                       NotifyFilters.Size;

            logFlushTimer.AutoReset = true;
            logFlushTimer.Interval = 5000; // sometimes the filesystem event does not trigger
            logFlushTimer.Elapsed += (o, e) => Task.Factory.StartNew(() => Update(false));
            logFlushTimer.Enabled = true;

            CurrentFile = LogEnumerator.GetLogFiles(LogDirectory).First();
            filePosition = new FileInfo(CurrentFile).Length;
            Update(false);
            fileWatcher.EnableRaisingEvents = true;
            logger.Debug("Started monitoring on folder {0}", LogDirectory);
        }

        private void FileWatcher_Created(object sender, FileSystemEventArgs e)
        {
            Update(checkOtherFiles: true);
        }

        private void FileWatcher_Changed(object sender, FileSystemEventArgs e)
        {
            if (Path.GetExtension(e.FullPath) != ".log")
                return;
            Update(checkOtherFiles: e.FullPath != CurrentFile);
        }

        private void Update(bool checkOtherFiles)
        {
            lock (@lock)
                try
                {
                    filePosition = ReadFileFromPosition(CurrentFile, filePosition);
                    if (checkOtherFiles) {
                        var latestFile = LogEnumerator.GetLogFiles(LogDirectory).First();
                        if (latestFile != CurrentFile)
                        {
                            CurrentFile = latestFile;
                            filePosition = ReadFileFromPosition(CurrentFile, 0);
                        }
                    }
                }
                catch (Exception e)
                {
                    logger.Error(e, "Error while reading log file, skipping");
                    filePosition = new FileInfo(CurrentFile).Length; // Skipping the 'poisoned' data
                }
        }

        private long ReadFileFromPosition(string file, long filePosition)
        {
            using (var fileReader = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
            {
                fileReader.Position = filePosition;
                using (var textReader = new StreamReader(fileReader))
                using (var jsonReader = new JsonTextReader(textReader) { SupportMultipleContent = true })
                {
                    JsonSerializer serializer = new JsonSerializer();
                    while (jsonReader.Read())
                    {
                        var record = serializer.Deserialize(jsonReader);
                        // Sometimes serializer gives out the json as string instead of JObject
                        var @object = record as JObject ?? JObject.Parse(record as string);
                        OnNext(LogEventConverter.Convert(@object));
                        logger.Debug("Received event: {0}", @object.ToString());
                    }
                    return fileReader.Position;
                }
            }
        }

        #region IDisposable Support
{'fraction_non_alphanumeric': 0.07439011049119353, 'fraction_numerical': 0.001969149983590417, 'mean_word_length': 1.7087407407407407, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4710869', 'n_tokens_mistral': 2139, 'n_tokens_neox': 2039, 'n_words': 535}