Dhcp Lease Queue worker

src/pdns-dhcp/Dhcp/DhcpLeaseQueue.cs

@@ -7,7 +7,23 @@ namespace pdns_dhcp.Dhcp;
 
 public class DhcpLeaseQueue
 {
-	private readonly Channel<DhcpLeaseChange> _pipe = Channel.CreateUnbounded<DhcpLeaseChange>();
+	private readonly Channel<DhcpLeaseChange> _pipe;
+	private readonly ChannelReader<DhcpLeaseChange> _reader;
+	private readonly ChannelWriter<DhcpLeaseChange> _writer;
+
+	public ref readonly ChannelReader<DhcpLeaseChange> Reader => ref _reader;
+
+	public DhcpLeaseQueue()
+	{
+		_pipe = Channel.CreateUnbounded<DhcpLeaseChange>();
+		_reader = _pipe.Reader;
+		_writer = _pipe.Writer;
+	}
+
+	public ValueTask Write(DhcpLeaseChange change, CancellationToken cancellationToken = default)
+	{
+		return _writer.WriteAsync(change, cancellationToken);
+	}
 }
 
 public readonly record struct DhcpLeaseChange(IPAddress Address, string FQDN, DhcpLeaseIdentifier Identifier, TimeSpan Lifetime)

src/pdns-dhcp/Dns/DnsRepository.cs

@@ -40,10 +40,10 @@ public class DnsRepository
 			cancellationToken, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
 	}
 
-	public async ValueTask Record(DhcpLeaseChange leaseChange)
+	public async ValueTask Record(DhcpLeaseChange leaseChange, CancellationToken cancellationToken = default)
 	{
 		// just lock that thing.
-		using (await _recordLock.AcquireLockAsync(CancellationToken.None).ConfigureAwait(false))
+		using (await _recordLock.AcquireLockAsync(cancellationToken).ConfigureAwait(false))
 		{
 			RecordContinuation(leaseChange);
 		}

src/pdns-dhcp/Kea/IKeaDhcpLeaseHandler.cs

@@ -1,8 +1,10 @@
 using nietras.SeparatedValues;
 
+using pdns_dhcp.Dhcp;
+
 namespace pdns_dhcp.Kea;
 
 public interface IKeaDhcpLeaseHandler
 {
-	void Handle(in SepReader.Row row);
+	DhcpLeaseChange? Handle(in SepReader.Row row);
 }

src/pdns-dhcp/Kea/IKeaFactory.cs

@@ -0,0 +1,14 @@
+using pdns_dhcp.Options;
+
+using Stl.Interception;
+
+namespace pdns_dhcp.Kea;
+
+public interface IKeaFactory : IRequiresFullProxy
+{
+	KeaDhcp4LeaseHandler CreateHandler4();
+
+	KeaDhcp6LeaseHandler CreateHandler6();
+
+	KeaDhcpLeaseWatcher CreateWatcher(IKeaDhcpLeaseHandler handler, KeaDhcpServerOptions options);
+}

src/pdns-dhcp/Kea/KeaDhcp4LeaseHandler.cs

@@ -1,11 +1,15 @@
 using nietras.SeparatedValues;
 
+using pdns_dhcp.Dhcp;
+
 namespace pdns_dhcp.Kea;
 
 public class KeaDhcp4LeaseHandler : IKeaDhcpLeaseHandler
 {
-	public void Handle(in SepReader.Row row)
+	public DhcpLeaseChange? Handle(in SepReader.Row row)
 	{
 		KeaDhcp4Lease lease = KeaDhcp4Lease.Parse(row);
+
+		return default;
 	}
 }

src/pdns-dhcp/Kea/KeaDhcp6LeaseHandler.cs

@@ -1,11 +1,15 @@
 using nietras.SeparatedValues;
 
+using pdns_dhcp.Dhcp;
+
 namespace pdns_dhcp.Kea;
 
 public class KeaDhcp6LeaseHandler : IKeaDhcpLeaseHandler
 {
-	public void Handle(in SepReader.Row row)
+	public DhcpLeaseChange? Handle(in SepReader.Row row)
 	{
 		KeaDhcp6Lease lease = KeaDhcp6Lease.Parse(row);
+
+		return default;
 	}
 }

src/pdns-dhcp/Kea/KeaDhcpLeaseWatcher.cs

@@ -7,36 +7,38 @@ using Microsoft.Extensions.Hosting;
 
 using nietras.SeparatedValues;
 
+using pdns_dhcp.Dhcp;
 using pdns_dhcp.Options;
 
 namespace pdns_dhcp.Kea;
 
-public sealed class KeaDhcpLeaseWatcher<T> : IHostedService
+public sealed class KeaDhcpLeaseWatcher : IHostedService
-	where T : IKeaDhcpLeaseHandler
 {
 	private static readonly FileStreamOptions LeaseFileStreamOptions = new()
 	{
 		Access = FileAccess.Read,
 		Mode = FileMode.Open,
-		Options = FileOptions.SequentialScan,
+		Options = FileOptions.SequentialScan | FileOptions.Asynchronous,
 		Share = (FileShare)7,
 	};
 
 	private readonly Decoder _decoder;
 	private readonly FileSystemWatcher _fsw;
-	private readonly T _handler;
+	private readonly IKeaDhcpLeaseHandler _handler;
 	private readonly string _leaseFile;
 	private readonly Pipe _pipe;
+	private readonly DhcpLeaseQueue _queue;
 	private Channel<FileSystemEventArgs>? _eventChannel;
 	private Task? _executeTask;
 	private CancellationTokenSource? _stoppingCts;
 
 	private KeaDhcpServerOptions Options { get; }
 
-	public KeaDhcpLeaseWatcher(KeaDhcpServerOptions options, T handler)
+	public KeaDhcpLeaseWatcher(KeaDhcpServerOptions options, IKeaDhcpLeaseHandler handler, DhcpLeaseQueue queue)
 	{
 		Options = options = options with { Leases = PathEx.ExpandPath(options.Leases) };
 		_handler = handler;
+		_queue = queue;
 
 		var leases = options.Leases.AsSpan();
 		if (leases.IsWhiteSpace())
@@ -177,7 +179,12 @@ public sealed class KeaDhcpLeaseWatcher<T> : IHostedService
 						return;
 					}
 
-					_handler.Handle(reader.Current);
+					if (_handler.Handle(reader.Current) is not { } lease)
+					{
+						continue;
+					}
+
+					await _queue.Write(lease, stoppingToken).ConfigureAwait(false);
 				}
 
 				var memory = writer.GetMemory();

src/pdns-dhcp/Kea/KeaService.cs

@@ -0,0 +1,50 @@
+using System.Collections.Immutable;
+
+using Microsoft.Extensions.Hosting;
+
+using pdns_dhcp.Options;
+
+namespace pdns_dhcp.Kea;
+
+public class KeaService : IHostedService
+{
+	private readonly ImmutableArray<IHostedService> _services;
+
+	public KeaService(KeaDhcpOptions options, IKeaFactory factory)
+	{
+		if (options.Dhcp4 is { } dhcp4Options)
+		{
+			_services.Add(factory.CreateWatcher(factory.CreateHandler4(), dhcp4Options));
+		}
+
+		if (options.Dhcp6 is { } dhcp6Options)
+		{
+			_services.Add(factory.CreateWatcher(factory.CreateHandler6(), dhcp6Options));
+		}
+	}
+
+	public Task StartAsync(CancellationToken cancellationToken = default)
+	{
+		Task[] tasks = new Task[_services.Length];
+		for (int i = 0; i < tasks.Length; i++)
+		{
+			tasks[i] = _services[i].StartAsync(cancellationToken);
+		}
+
+		return Task.WhenAll(tasks);
+	}
+
+	public async Task StopAsync(CancellationToken cancellationToken = default)
+	{
+		Task[] tasks = new Task[_services.Length];
+		for (int i = 0; i < tasks.Length; i++)
+		{
+			tasks[i] = _services[i].StopAsync(cancellationToken);
+		}
+
+		var waitTask = Task.WhenAll(tasks);
+		TaskCompletionSource taskCompletionSource = new();
+		using var registration = cancellationToken.Register(s => ((TaskCompletionSource)s!).SetCanceled(), taskCompletionSource);
+		await Task.WhenAny(waitTask, taskCompletionSource.Task).ConfigureAwait(continueOnCapturedContext: false);
+	}
+}

src/pdns-dhcp/Program.cs

@@ -9,6 +9,8 @@ using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 
 using pdns_dhcp.Connections;
+using pdns_dhcp.Dhcp;
+using pdns_dhcp.Dns;
 using pdns_dhcp.Kea;
 using pdns_dhcp.Options;
 using pdns_dhcp.PowerDns;
@@ -21,13 +23,14 @@ var builder = WebApplication.CreateBuilder(args);
 builder.Services.Configure<DhcpOptions>(builder.Configuration.GetRequiredSection("Dhcp"));
 builder.Services.Configure<PowerDnsOptions>(builder.Configuration.GetRequiredSection("PowerDns"));
 
-builder.Services.AddHostedService<DhcpLeaseWatcher>();
+builder.Services.AddHostedService<DhcpWatcher>();
-builder.Services.AddHostedService<PowerDnsBackend>();
+builder.Services.AddHostedService<DnsQueueWorker>();
 
-builder.Services.AddTypedFactory<IDhcpLeaseWatcherFactory>();
+builder.Services.AddSingleton<DhcpLeaseQueue>();
+builder.Services.AddSingleton<DnsRepository>();
 
-builder.Services.AddTransient<KeaDhcp4LeaseHandler>();
+builder.Services.AddTypedFactory<IDhcpWatcherFactory>();
-builder.Services.AddTransient<KeaDhcp6LeaseHandler>();
+builder.Services.AddTypedFactory<IKeaFactory>();
 
 builder.Services.Configure<SocketTransportOptions>(options =>
 {

src/pdns-dhcp/Services/DhcpWatcher.cs

@@ -7,31 +7,23 @@ using pdns_dhcp.Options;
 
 namespace pdns_dhcp.Services;
 
-public class DhcpLeaseWatcher : IHostedService
+public class DhcpWatcher : IHostedService
 {
 	private readonly ImmutableArray<IHostedService> _services;
 
-	public DhcpLeaseWatcher(IOptions<DhcpOptions> options, IDhcpLeaseWatcherFactory factory)
+	public DhcpWatcher(IOptions<DhcpOptions> options, IDhcpWatcherFactory factory)
 	{
 		var dhcpOptions = options.Value;
 		var services = ImmutableArray.CreateBuilder<IHostedService>();
 		if (dhcpOptions.Kea is { } keaOptions)
 		{
-			if (keaOptions.Dhcp4 is { } dhcp4Options)
+			services.Add(factory.KeaService(keaOptions));
-			{
-				services.Add(factory.KeaDhcp4Watcher(dhcp4Options));
-			}
-
-			if (keaOptions.Dhcp6 is { } dhcp6Options)
-			{
-				services.Add(factory.KeaDhcp6Watcher(dhcp6Options));
-			}
 		}
 
 		_services = services.DrainToImmutable();
 	}
 
-	public Task StartAsync(CancellationToken cancellationToken)
+	public Task StartAsync(CancellationToken cancellationToken = default)
 	{
 		Task[] tasks = new Task[_services.Length];
 		for (int i = 0; i < tasks.Length; i++)
@@ -42,7 +34,7 @@ public class DhcpLeaseWatcher : IHostedService
 		return Task.WhenAll(tasks);
 	}
 
-	public async Task StopAsync(CancellationToken cancellationToken)
+	public async Task StopAsync(CancellationToken cancellationToken = default)
 	{
 		Task[] tasks = new Task[_services.Length];
 		for (int i = 0; i < tasks.Length; i++)

src/pdns-dhcp/Services/DnsQueueWorker.cs

@@ -0,0 +1,31 @@
+using System.Threading.Channels;
+
+using Microsoft.Extensions.Hosting;
+
+using pdns_dhcp.Dhcp;
+using pdns_dhcp.Dns;
+
+namespace pdns_dhcp.Services;
+
+public class DnsQueueWorker : BackgroundService
+{
+	private readonly ChannelReader<DhcpLeaseChange> _channelReader;
+	private readonly DnsRepository _repository;
+
+	public DnsQueueWorker(DhcpLeaseQueue queue, DnsRepository repository)
+	{
+		_channelReader = queue.Reader;
+		_repository = repository;
+	}
+
+	protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+	{
+		while (await _channelReader.WaitToReadAsync(stoppingToken).ConfigureAwait(false))
+		{
+			while (_channelReader.TryRead(out var lease))
+			{
+				await _repository.Record(default, stoppingToken).ConfigureAwait(false);
+			}
+		}
+	}
+}

src/pdns-dhcp/Services/IDhcpLeaseWatcherFactory.cs

@@ -1,13 +0,0 @@
-using pdns_dhcp.Kea;
-using pdns_dhcp.Options;
-
-using Stl.Interception;
-
-namespace pdns_dhcp.Services;
-
-public interface IDhcpLeaseWatcherFactory : IRequiresFullProxy
-{
-	KeaDhcpLeaseWatcher<KeaDhcp4LeaseHandler> KeaDhcp4Watcher(KeaDhcpServerOptions options);
-
-	KeaDhcpLeaseWatcher<KeaDhcp6LeaseHandler> KeaDhcp6Watcher(KeaDhcpServerOptions options);
-}

src/pdns-dhcp/Services/IDhcpWatcherFactory.cs

@@ -0,0 +1,11 @@
+using pdns_dhcp.Kea;
+using pdns_dhcp.Options;
+
+using Stl.Interception;
+
+namespace pdns_dhcp.Services;
+
+public interface IDhcpWatcherFactory : IRequiresFullProxy
+{
+	KeaService KeaService(KeaDhcpOptions options);
+}

src/pdns-dhcp/Services/PowerDnsBackend.cs

@@ -1,31 +0,0 @@
-using Microsoft.Extensions.Hosting;
-
-namespace pdns_dhcp.Services;
-
-public class PowerDnsBackend : BackgroundService
-{
-	public PowerDnsBackend()
-	{
-	}
-
-	~PowerDnsBackend()
-	{
-		DisposeCore();
-	}
-
-	protected override Task ExecuteAsync(CancellationToken stoppingToken)
-	{
-		return Task.CompletedTask;		
-	}
-
-	public override void Dispose()
-	{
-		base.Dispose();
-		DisposeCore();
-		GC.SuppressFinalize(this);
-	}
-
-	private void DisposeCore()
-	{
-	}
-}