Refactor integrated server management

app/Desktop/Desktop.csproj

@@ -21,7 +21,6 @@
     <PackageReference Include="Avalonia.Desktop" Version="11.0.6" />
     <PackageReference Include="Avalonia.Diagnostics" Version="11.0.6" Condition=" '$(Configuration)' == 'Debug' " />
     <PackageReference Include="Avalonia.Fonts.Inter" Version="11.0.6" />
-    <PackageReference Include="Avalonia.ReactiveUI" Version="11.0.6" />
     <PackageReference Include="Avalonia.Themes.Fluent" Version="11.0.6" />
   </ItemGroup>
   

app/Desktop/Main/Pages/AttachmentsPageModel.cs

@@ -1,9 +1,9 @@
 using System;
 using System.Collections.Generic;
 using System.ComponentModel;
-using System.Reactive.Linq;
+using System.Threading;
 using System.Threading.Tasks;
-using Avalonia.ReactiveUI;
+using Avalonia.Threading;
 using DHT.Desktop.Common;
 using DHT.Desktop.Main.Controls;
 using DHT.Server;
@@ -11,6 +11,7 @@ using DHT.Server.Data;
 using DHT.Server.Data.Aggregations;
 using DHT.Server.Data.Filters;
 using DHT.Server.Database;
+using DHT.Server.Download;
 using DHT.Utils.Models;
 using DHT.Utils.Tasks;
 
@@ -60,7 +61,6 @@ sealed class AttachmentsPageModel : BaseModel, IDisposable {
 	private readonly State state;
 	private readonly AsyncValueComputer<DownloadStatusStatistics>.Single downloadStatisticsComputer;
 
-	private IDisposable? finishedItemsSubscription;
 	private int doneItemsCount;
 	private int initialFinishedCount;
 	private int? totalItemsToDownloadCount;
@@ -76,11 +76,12 @@ sealed class AttachmentsPageModel : BaseModel, IDisposable {
 		downloadStatisticsComputer.Recompute();
 
 		state.Db.Statistics.PropertyChanged += OnDbStatisticsChanged;
+		state.Downloader.OnItemFinished += DownloaderOnOnItemFinished;
 	}
 
 	public void Dispose() {
 		state.Db.Statistics.PropertyChanged -= OnDbStatisticsChanged;
-		finishedItemsSubscription?.Dispose();
+		state.Downloader.OnItemFinished -= DownloaderOnOnItemFinished;
 		FilterModel.Dispose();
 	}
 
@@ -138,22 +139,19 @@ sealed class AttachmentsPageModel : BaseModel, IDisposable {
 		OnPropertyChanged(nameof(DownloadProgress));
 	}
 
-	private void OnItemsFinished(int finishedItemCount) {
-		doneItemsCount += finishedItemCount;
-		UpdateDownloadMessage();
+	private void DownloaderOnOnItemFinished(object? sender, DownloadItem e) {
+		Interlocked.Increment(ref doneItemsCount);
+		
+		Dispatcher.UIThread.Invoke(UpdateDownloadMessage);
 		downloadStatisticsComputer.Recompute();
 	}
 
 	public async Task OnClickToggleDownload() {
-		IsToggleDownloadButtonEnabled = false;
-		
 		if (IsDownloading) {
+			IsToggleDownloadButtonEnabled = false;
 			await state.Downloader.Stop();
-			
-			finishedItemsSubscription?.Dispose();
-			finishedItemsSubscription = null;
-			
 			downloadStatisticsComputer.Recompute();
+			IsToggleDownloadButtonEnabled = true;
 
 			state.Db.RemoveDownloadItems(EnqueuedItemFilter, FilterRemovalMode.RemoveMatching);
 
@@ -163,22 +161,13 @@ sealed class AttachmentsPageModel : BaseModel, IDisposable {
 			UpdateDownloadMessage();
 		}
 		else {
-			var finishedItems = await state.Downloader.Start();
-
 			initialFinishedCount = statisticsDownloaded.Items + statisticsFailed.Items;
-			finishedItemsSubscription = finishedItems.Select(static _ => true)
-			                                         .Buffer(TimeSpan.FromMilliseconds(100))
-			                                         .Select(static items => items.Count)
-			                                         .Where(static items => items > 0)
-			                                         .ObserveOn(AvaloniaScheduler.Instance)
-			                                         .Subscribe(OnItemsFinished);
-			
+			await state.Downloader.Start();
 			EnqueueDownloadItems();
 		}
 
 		OnPropertyChanged(nameof(ToggleDownloadButtonText));
 		OnPropertyChanged(nameof(IsDownloading));
-		IsToggleDownloadButtonEnabled = true;
 	}
 
 	public void OnClickRetryFailedDownloads() {

app/Server/Download/Downloader.cs

@@ -9,6 +9,8 @@ public sealed class Downloader {
 	private DownloaderTask? current;
 	public bool IsDownloading => current != null;
 	
+	public event EventHandler<DownloadItem>? OnItemFinished;
+	
 	private readonly IDatabaseFile db;
 	private readonly SemaphoreSlim semaphore = new (1, 1);
 	
@@ -16,11 +18,13 @@ public sealed class Downloader {
 		this.db = db;
 	}
 
-	public async Task<IObservable<DownloadItem>> Start() {
+	public async Task Start() {
 		await semaphore.WaitAsync();
 		try {
-			current ??= new DownloaderTask(db);
-			return current.FinishedItems;
+			if (current == null) {
+				current = new DownloaderTask(db);
+				current.OnItemFinished += DelegateOnItemFinished;
+			}
 		} finally {
 			semaphore.Release();
 		}
@@ -30,11 +34,16 @@ public sealed class Downloader {
 		await semaphore.WaitAsync();
 		try {
 			if (current != null) {
-				await current.DisposeAsync();
+				await current.Stop();
+				current.OnItemFinished -= DelegateOnItemFinished;
 				current = null;
 			}
 		} finally {
 			semaphore.Release();
 		}
 	}
+
+	private void DelegateOnItemFinished(object? sender, DownloadItem e) {
+		OnItemFinished?.Invoke(this, e);
+	}
 }

app/Server/Download/DownloaderTask.cs

@@ -1,23 +1,25 @@
 using System;
 using System.Linq;
 using System.Net.Http;
-using System.Reactive.Subjects;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using DHT.Server.Database;
 using DHT.Utils.Logging;
+using DHT.Utils.Models;
 using DHT.Utils.Tasks;
 
 namespace DHT.Server.Download;
 
-sealed class DownloaderTask : IAsyncDisposable {
+sealed class DownloaderTask : BaseModel {
 	private static readonly Log Log = Log.ForType<DownloaderTask>();
 
 	private const int DownloadTasks = 4;
 	private const int QueueSize = 25;
 	private const string UserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36";
 
+	internal event EventHandler<DownloadItem>? OnItemFinished;
+
 	private readonly Channel<DownloadItem> downloadQueue = Channel.CreateBounded<DownloadItem>(new BoundedChannelOptions(QueueSize) {
 		SingleReader = false,
 		SingleWriter = true,
@@ -29,16 +31,12 @@ sealed class DownloaderTask : IAsyncDisposable {
 	private readonly CancellationToken cancellationToken;
 
 	private readonly IDatabaseFile db;
-	private readonly Subject<DownloadItem> finishedItemPublisher = new ();
-	
 	private readonly Task queueWriterTask;
 	private readonly Task[] downloadTasks;
 
-	public IObservable<DownloadItem> FinishedItems => finishedItemPublisher;
-
 	internal DownloaderTask(IDatabaseFile db) {
-		this.db = db;
 		this.cancellationToken = cancellationTokenSource.Token;
+		this.db = db;
 		this.queueWriterTask = Task.Run(RunQueueWriterTask);
 		this.downloadTasks = Enumerable.Range(1, DownloadTasks).Select(taskIndex => Task.Run(() => RunDownloadTask(taskIndex))).ToArray();
 	}
@@ -81,7 +79,7 @@ sealed class DownloaderTask : IAsyncDisposable {
 				log.Error(e);
 			} finally {
 				try {
-					finishedItemPublisher.OnNext(item);
+					OnItemFinished?.Invoke(this, item);
 				} catch (Exception e) {
 					log.Error("Caught exception in event handler: " + e);
 				}
@@ -89,7 +87,7 @@ sealed class DownloaderTask : IAsyncDisposable {
 		}
 	}
 
-	public async ValueTask DisposeAsync() {
+	internal async Task Stop() {
 		try {
 			await cancellationTokenSource.CancelAsync();
 		} catch (Exception) {
@@ -104,7 +102,6 @@ sealed class DownloaderTask : IAsyncDisposable {
 			await Task.WhenAll(downloadTasks).WaitIgnoringCancellation();
 		} finally {
 			cancellationTokenSource.Dispose();
-			finishedItemPublisher.OnCompleted();
 		}
 	}
 }

app/Server/Server.csproj

@@ -12,7 +12,6 @@
   
   <ItemGroup>
     <PackageReference Include="Microsoft.Data.Sqlite" Version="8.0.0" />
-    <PackageReference Include="System.Reactive" Version="6.0.0" />
   </ItemGroup>
   
   <ItemGroup>

app/Server/State.cs

@@ -21,7 +21,6 @@ public sealed class State : IAsyncDisposable {
 
 	public async ValueTask DisposeAsync() {
 		await Downloader.Stop();
-		await Server.Stop();
 		Db.Dispose();
 	}
 }