diff --git a/.gitignore b/.gitignore index a238d01..29e0cca 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ robin/bin/* robin/obj/* robin/.env +.vs/* diff --git a/robin/Configuration/BuilderExtensions.cs b/robin/Configuration/BuilderExtensions.cs index e8b65a0..9b1cc42 100644 --- a/robin/Configuration/BuilderExtensions.cs +++ b/robin/Configuration/BuilderExtensions.cs @@ -10,7 +10,9 @@ public static class BuilderExtensions this IHostApplicationBuilder builder ) { - DotNetEnv.Env.Load(); +#if DEBUG + DotNetEnv.Env.Load(Path.Combine(AppContext.BaseDirectory, ".env")); +#endif builder.ThrowIfNull().Configuration.AddEnvironmentVariables(); return builder; } @@ -20,7 +22,7 @@ public static class BuilderExtensions builder .ThrowIfNull() .Configuration.AddJsonFile( - path: "appsettings.json", + path: Path.Combine(AppContext.BaseDirectory, "appsettings.json"), optional: false, reloadOnChange: true ); diff --git a/robin/Options/BuilderExtensions.cs b/robin/Options/BuilderExtensions.cs index 34eef6a..5ffb88f 100644 --- a/robin/Options/BuilderExtensions.cs +++ b/robin/Options/BuilderExtensions.cs @@ -9,7 +9,15 @@ public static class BuilderExtensions { builder .Services.AddOptions() - .Bind(builder.Configuration.GetSection("Monitor")); + .Bind(builder.Configuration.GetSection(MonitorOptions.SectionName)); + + builder + .Services.AddOptions() + .Bind(builder.Configuration.GetSection(ElasticOptions.SectionName)); + + builder + .Services.AddOptions() + .Bind(builder.Configuration.GetSection(PositionPersistenceOptions.SectionName)); return builder; } diff --git a/robin/Options/ElasticOptions.cs b/robin/Options/ElasticOptions.cs new file mode 100644 index 0000000..344c7c2 --- /dev/null +++ b/robin/Options/ElasticOptions.cs @@ -0,0 +1,11 @@ +namespace Robin; + +public sealed class ElasticOptions +{ + public const string SectionName = "Elastic"; + + public Uri? Uri { get; set; } + public string? Index { get; set; } + public string? Username { get; set; } + public string? Password { get; set; } +} diff --git a/robin/Options/MonitorOptions.cs b/robin/Options/MonitorOptions.cs index 160d4a0..dca53ac 100644 --- a/robin/Options/MonitorOptions.cs +++ b/robin/Options/MonitorOptions.cs @@ -1,6 +1,8 @@ namespace Robin.Options; -public sealed class MonitorOptions() +public sealed class MonitorOptions { - public required List Paths { get; init; } = []; + public const string SectionName = "Monitor"; + + public List Paths { get; set; } = []; } diff --git a/robin/Options/PositionPersistenceOptions.cs b/robin/Options/PositionPersistenceOptions.cs new file mode 100644 index 0000000..30b6d17 --- /dev/null +++ b/robin/Options/PositionPersistenceOptions.cs @@ -0,0 +1,8 @@ +namespace Robin.Options; + +public sealed class PositionPersistenceOptions +{ + public const string SectionName = "PositionPersistence"; + + public string? FilePath { get; set; } +} diff --git a/robin/Robin.csproj b/robin/Robin.csproj index a9cb775..f92a728 100644 --- a/robin/Robin.csproj +++ b/robin/Robin.csproj @@ -15,6 +15,7 @@ + diff --git a/robin/Serialization/SerializationContext.cs b/robin/Serialization/SerializationContext.cs new file mode 100644 index 0000000..8758ea8 --- /dev/null +++ b/robin/Serialization/SerializationContext.cs @@ -0,0 +1,7 @@ +using System.Collections.Concurrent; +using System.Text.Json.Serialization; + +namespace Robin.Serialization; + +[JsonSerializable(typeof(ConcurrentDictionary))] +public partial class SerializationContext : JsonSerializerContext { } diff --git a/robin/Services/BuilderExtensions.cs b/robin/Services/BuilderExtensions.cs index 4c1f566..5e49784 100644 --- a/robin/Services/BuilderExtensions.cs +++ b/robin/Services/BuilderExtensions.cs @@ -1,6 +1,11 @@ using System.Core.Extensions; +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Nest; +using Robin.Services.Monitoring; +using Robin.Services.Persistence; +using Robin.Services.Publishing; namespace Robin.Services; @@ -8,9 +13,31 @@ public static class BuilderExtensions { public static IHostApplicationBuilder AddRobinServices(this IHostApplicationBuilder builder) { + var elasticClientOptions = builder + .Configuration.GetSection(ElasticOptions.SectionName) + .Get(); + + if (elasticClientOptions is null) + { + throw new InvalidOperationException( + "ElasticOptions section is missing in configuration." + ); + } + builder.ThrowIfNull(); builder.Services.AddHostedService(); - + builder.Services.AddSingleton( + new ElasticClient( + new ConnectionSettings(elasticClientOptions.Uri) + .BasicAuthentication( + elasticClientOptions.Username, + elasticClientOptions.Password + ) + .DefaultIndex(elasticClientOptions.Index) + ) + ); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); return builder; } } diff --git a/robin/Services/MonitoringService.cs b/robin/Services/Monitoring/MonitoringService.cs similarity index 54% rename from robin/Services/MonitoringService.cs rename to robin/Services/Monitoring/MonitoringService.cs index e38a7b3..7985411 100644 --- a/robin/Services/MonitoringService.cs +++ b/robin/Services/Monitoring/MonitoringService.cs @@ -1,19 +1,24 @@ +using System.Extensions; using System.Extensions.Core; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Robin.Options; +using Robin.Services.Persistence; -namespace Robin.Services; +namespace Robin.Services.Monitoring; public sealed class MonitoringService( + PositionPersistenceService persistenceService, IOptions options, ILogger logger ) : IHostedLifecycleService { + private readonly PositionPersistenceService persistenceService = persistenceService; private readonly MonitorOptions options = options.Value; + private readonly CancellationTokenSource cancellationTokenSource = new(); private readonly ILogger logger = logger; - private readonly List watchers = []; + private readonly List watchers = []; public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; @@ -22,12 +27,18 @@ public sealed class MonitoringService( public Task StartingAsync(CancellationToken cancellationToken) => Task.CompletedTask; - public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task StopAsync(CancellationToken cancellationToken) => this.CancelMonitors(); public Task StoppedAsync(CancellationToken cancellationToken) => Task.CompletedTask; public Task StoppingAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task CancelMonitors() + { + this.cancellationTokenSource.Cancel(); + return Task.WhenAll(this.watchers); + } + public async Task MonitorFiles(CancellationToken cancellationToken) { var scopedLogger = this.logger.CreateScopedLogger(); @@ -41,65 +52,56 @@ public sealed class MonitoringService( string.Join(", ", expandedPaths) ); + var cts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + this.cancellationTokenSource.Token + ); foreach (var filePath in expandedPaths) { - var watcher = CreateWatcher(filePath); - if (watcher is null) - { - scopedLogger.LogWarning("Failed to watch file: {file}", filePath); - continue; - } - - this.watchers.Add(watcher); - scopedLogger.LogDebug("Watching file: {file}", filePath); + this.watchers.Add(Task.Run(() => this.MonitorFile(filePath, cts.Token), cts.Token)); } } - private FileSystemWatcher? CreateWatcher(string filePath) - { - var scpoedLogger = this.logger.CreateScopedLogger(); - var directory = Path.GetDirectoryName(filePath); - var fileName = Path.GetFileName(filePath); - if (directory is null || fileName is null) - { - scpoedLogger.LogWarning("Invalid file path: {filePath}", filePath); - return default; - } - - var watcher = new FileSystemWatcher(directory, fileName) - { - NotifyFilter = NotifyFilters.LastWrite | NotifyFilters.Size, - EnableRaisingEvents = true, - }; - - watcher.Changed += this.OnFileChanged; - watcher.Error += this.OnWatcherError; - return watcher; - } - - private void OnFileChanged(object sender, FileSystemEventArgs e) + private async Task MonitorFile(string filePath, CancellationToken cancellationToken) { var scopedLogger = this.logger.CreateScopedLogger(); - scopedLogger.LogDebug("File changed: {file}", e.FullPath); - // TODO:: Read changes and push to Elastic - } - - private void OnWatcherError(object sender, ErrorEventArgs e) - { - var scopedLogger = this.logger.CreateScopedLogger(); - scopedLogger.LogError(e.GetException(), "Watcher error"); - } - - public void Dispose() - { - foreach (var watcher in this.watchers) + if (!File.Exists(filePath)) { - watcher.Changed -= this.OnFileChanged; - watcher.Error -= this.OnWatcherError; - watcher.Dispose(); + scopedLogger.LogWarning("File does not exist: {filePath}", filePath); + return; } - this.watchers.Clear(); + scopedLogger.LogInformation("Starting to monitor file: {filePath}", filePath); + var previousPosition = this.persistenceService.GetPosition(filePath); + using var fs = new FileStream( + filePath, + FileMode.Open, + FileAccess.Read, + FileShare.ReadWrite | FileShare.Delete + ); + + if (fs.Length < previousPosition) + { + previousPosition = 0; + } + + if (previousPosition > 0) + { + fs.Seek(previousPosition, SeekOrigin.Begin); + } + + using var reader = new StreamReader(fs); + var line = string.Empty; + while ((line = await reader.ReadLineAsync(cancellationToken))?.IsNullOrEmpty() is false) + { + scopedLogger.LogDebug("Read line: {line}", line); + //TODO: Publish to IPublishingService + await this.persistenceService.SavePositionAsync( + filePath, + fs.Position, + cancellationToken + ); + } } private static List ExpandPaths(List? patterns) diff --git a/robin/Services/Persistence/PositionPersistenceService.cs b/robin/Services/Persistence/PositionPersistenceService.cs new file mode 100644 index 0000000..9099df6 --- /dev/null +++ b/robin/Services/Persistence/PositionPersistenceService.cs @@ -0,0 +1,64 @@ +using System.Collections.Concurrent; +using System.Core.Extensions; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Robin.Options; +using Robin.Serialization; + +namespace Robin.Services.Persistence; + +public sealed class PositionPersistenceService +{ + private readonly PositionPersistenceOptions options; + private readonly ConcurrentDictionary positions = new(); + + public PositionPersistenceService( + IOptions options, + ILogger logger + ) + { + this.options = options.Value; + this.options.FilePath.ThrowIfNull(); + + try + { + if (File.Exists(this.options.FilePath)) + { + this.positions = + JsonSerializer.Deserialize( + File.ReadAllText(this.options.FilePath), + SerializationContext.Default.ConcurrentDictionaryStringInt64 + ) ?? new ConcurrentDictionary(); + } + } + catch (Exception ex) + { + logger.LogError(ex, "Failed to load positions from file."); + } + + this.positions ??= []; + } + + public async ValueTask SavePositionAsync( + string source, + long position, + CancellationToken cancellationToken + ) + { + this.positions[source] = position; + await File.WriteAllTextAsync( + this.options.FilePath.ThrowIfNull(), + JsonSerializer.Serialize( + this.positions, + SerializationContext.Default.ConcurrentDictionaryStringInt64 + ), + cancellationToken + ); + } + + public long GetPosition(string source) + { + return this.positions.GetOrAdd(source, 0); + } +} diff --git a/robin/Services/Publishing/ElasticPublishingService.cs b/robin/Services/Publishing/ElasticPublishingService.cs new file mode 100644 index 0000000..e46a611 --- /dev/null +++ b/robin/Services/Publishing/ElasticPublishingService.cs @@ -0,0 +1,35 @@ +using System.Extensions; +using System.Extensions.Core; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Nest; + +namespace Robin.Services.Publishing; + +public sealed class ElasticPublishingService( + IElasticClient elasticClient, + IOptions options, + ILogger logger +) : IPublishingService +{ + private readonly IElasticClient elasticClient = elasticClient; + private readonly ElasticOptions options = options.Value; + private readonly ILogger logger = logger; + + public async Task PublishAsync(T item, CancellationToken cancellationToken) + where T : class + { + var response = await this.elasticClient.IndexDocumentAsync(item, cancellationToken); + if (response.IsValid) + { + return true; + } + + var scopedLogger = this.logger.CreateScopedLogger(); + scopedLogger.LogError( + "Failed to publish document to Elasticsearch. Error: {Error}", + response.DebugInformation + ); + return false; + } +} diff --git a/robin/Services/Publishing/IPublishingService.cs b/robin/Services/Publishing/IPublishingService.cs new file mode 100644 index 0000000..e1abcef --- /dev/null +++ b/robin/Services/Publishing/IPublishingService.cs @@ -0,0 +1,7 @@ +namespace Robin.Services.Publishing; + +public interface IPublishingService +{ + Task PublishAsync(T item, CancellationToken cancellationToken) + where T : class; +} diff --git a/robin/appsettings.json b/robin/appsettings.json index 0a4956c..ce00c33 100644 --- a/robin/appsettings.json +++ b/robin/appsettings.json @@ -8,5 +8,8 @@ "Robin.Services": "Debug" } } + }, + "PositionPersistence": { + "FilePath": "positions.json" } }