using System.Collections.ObjectModel;
using System.Diagnostics;
using System.Globalization;
using System.Text;
using AvParser.Core.Parsing;
using AvParser.Core.Proxies;
using AvParser.Core.Settings;
using AvParser.UI.Localization;
using AvParser.UI.Navigation;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using ReactiveUI;
using ReactiveUI.Primitives;
using ReactiveUI.Primitives.Concurrency;
using ReactiveUI.Primitives.Signals;
using ReactiveUI.SourceGenerators;
namespace AvParser.UI.ViewModels;
/// Runs a parser over pasted text and streams the results into the UI.
///
/// This page exists to exercise the whole contract —
/// streaming, progress and cancellation — rather than to be a finished feature.
///
public partial class ParseViewModel : PageViewModel, IDisposable
{
/// Records buffered before being pushed to the UI collection in one go.
private const int BatchSize = 512;
///
/// Upper bound on rows shown. Beyond this the parse still completes and the count stays
/// accurate, but the list stops growing — truncation is reported, never silent.
///
private const int MaxDisplayedRecords = 20_000;
private readonly IParserCatalog _catalog;
private readonly ISettingsService _settings;
private readonly IProxyPool _proxyPool;
private readonly IServiceProvider _services;
private readonly ILogger _logger;
private readonly ISequencer _mainThread;
private readonly ObservableAsPropertyHelper _isBusy;
private readonly Signal _proxyChanged = new();
private CancellationTokenSource? _cancellation;
/// Text to parse.
[Reactive]
public partial string InputText { get; set; }
/// Parser applied by .
[Reactive]
public partial ParserViewModel SelectedParser { get; set; }
/// Completion of the running parse, 0.0 to 1.0.
[Reactive]
public partial double Progress { get; set; }
/// Outcome summary shown under the toolbar; when idle.
[Reactive]
public partial string? StatusMessage { get; set; }
///
/// Whether the selected parser needs the network but has no working proxy to use.
///
///
/// Only network parsers are gated. A parser that works on text the user pasted has nothing to
/// route, and blocking it would make the app unusable whenever the public lists are down.
///
[Reactive]
public partial bool IsBlockedWithoutProxy { get; set; }
/// Creates the page.
/// Available parsers.
/// Used to remember the selected parser.
/// Consulted for the live count that gates network parsers.
/// Resolves the navigation service lazily, to keep pages acyclic.
/// Diagnostics.
///
/// Scheduler used to marshal collection and progress updates back to the UI thread. Tests
/// pass to make everything synchronous.
///
public ParseViewModel(
IParserCatalog catalog,
ISettingsService settings,
IProxyPool proxyPool,
IServiceProvider services,
ILogger logger,
ISequencer? mainThread = null
)
{
_catalog = catalog ?? throw new ArgumentNullException(nameof(catalog));
_settings = settings ?? throw new ArgumentNullException(nameof(settings));
_proxyPool = proxyPool ?? throw new ArgumentNullException(nameof(proxyPool));
_services = services ?? throw new ArgumentNullException(nameof(services));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_mainThread = mainThread ?? RxSchedulers.MainThreadScheduler;
InputText = string.Empty;
Parsers = [.. catalog.Parsers.Select(parser => new ParserViewModel(parser))];
SelectedParser = Parsers.First(parser => parser.Id == catalog.FindOrDefault(settings.Current.LastParserId).Id);
// The pool changes on every lease outcome and on every probe, so coalesce before
// re-evaluating whether the parser is allowed to run.
_proxyPool.Changed += OnProxyPoolChanged;
_proxyChanged
.Throttle(TimeSpan.FromMilliseconds(250), _mainThread)
.ObserveOn(_mainThread)
.Subscribe(_ => RefreshProxyGate());
RefreshProxyGate();
var canParse = this.WhenAnyValue(
x => x.InputText,
x => x.IsBlockedWithoutProxy,
static (text, blocked) => (text, blocked)
)
.Select(static state => !string.IsNullOrWhiteSpace(state.text) && !state.blocked)
.DistinctUntilChanged();
ParseCommand = ReactiveCommand.CreateFromTask(RunParseAsync, canParse, _mainThread);
GoToProxiesCommand = ReactiveCommand.Create(
() => _services.GetRequiredService().NavigateTo(),
outputScheduler: _mainThread
);
_isBusy = ParseCommand.IsExecuting.ToProperty(this, nameof(IsBusy), false, _mainThread);
CancelCommand = ReactiveCommand.Create(() => _cancellation?.Cancel(), ParseCommand.IsExecuting, _mainThread);
ClearCommand = ReactiveCommand.Create(
() =>
{
InputText = string.Empty;
ClearResults();
StatusMessage = null;
Progress = 0d;
},
ParseCommand.IsExecuting.Select(static running => !running),
_mainThread
);
LoadSampleCommand = ReactiveCommand.Create(
() => InputText = SampleFor(SelectedParser.Id),
ParseCommand.IsExecuting.Select(static running => !running),
_mainThread
);
GenerateLargeSampleCommand = ReactiveCommand.Create(
() => InputText = LargeSampleFor(SelectedParser.Id),
ParseCommand.IsExecuting.Select(static running => !running),
_mainThread
);
// Remember the parser choice; the debounced settings service coalesces the writes.
// Switching parser can also change whether the gate applies, since only network parsers
// are gated.
this.WhenAnyValue(x => x.SelectedParser)
.Where(static parser => parser is not null)
.Subscribe(parser =>
{
_settings.Update(current => current with { LastParserId = parser.Id });
RefreshProxyGate();
});
_settings.Changes.Subscribe(_ => RefreshProxyGate());
// Errors surfacing from any command must not tear the process down.
ParseCommand.ThrownExceptions.Subscribe(OnCommandFailed);
}
///
public override string TitleKey => "Page.Parse";
///
public override string IconKey => "IconDocument";
/// Every registered parser, for the picker.
public IReadOnlyList Parsers { get; }
/// Successfully parsed records, capped at .
public ObservableCollection Records { get; } = [];
/// Per-line failures. A failure never aborts the parse.
public ObservableCollection Errors { get; } = [];
/// Whether a parse is currently running.
public bool IsBusy => _isBusy.Value;
/// Runs over .
public ReactiveCommand ParseCommand { get; }
/// Cancels the running parse.
public ReactiveCommand CancelCommand { get; }
/// Clears the input and all results.
public ReactiveCommand ClearCommand { get; }
/// Fills the input with a small example for the selected parser.
public ReactiveCommand LoadSampleCommand { get; }
/// Fills the input with 50 000 rows, so progress and cancellation are observable.
public ReactiveCommand GenerateLargeSampleCommand { get; }
/// Takes the user to the page where the proxy problem can be fixed.
public ReactiveCommand GoToProxiesCommand { get; }
/// Explains why parsing is blocked.
public string ProxyRequiredMessage => Localizer.Instance["Parse.ProxyRequired"];
/// Re-evaluates the proxy gate. Exposed so tests can drive it without waiting.
public void RefreshProxyGate()
{
var settings = _settings.Current;
IsBlockedWithoutProxy =
SelectedParser.Parser.RequiresNetwork && !settings.AllowDirectConnection && _proxyPool.LiveCount == 0;
}
private void OnProxyPoolChanged(object? sender, EventArgs e) => _proxyChanged.OnNext(RxVoid.Default);
private async Task RunParseAsync(CancellationToken commandToken)
{
using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(commandToken);
_cancellation = cancellation;
var parser = SelectedParser.Parser;
var input = InputText;
var token = cancellation.Token;
ClearResults();
Progress = 0d;
StatusMessage = null;
var recordBuffer = new List(BatchSize);
var errorBuffer = new List(16);
var progress = new Progress(value => OnUi(() => Progress = value.Fraction));
var stopwatch = Stopwatch.StartNew();
var succeeded = 0;
var failed = 0;
var truncated = false;
var cancelled = false;
try
{
await foreach (var outcome in parser.ParseAsync(input, progress, token).ConfigureAwait(false))
{
if (outcome.IsSuccess)
{
succeeded++;
if (succeeded <= MaxDisplayedRecords)
{
recordBuffer.Add(outcome.Value!);
}
else
{
truncated = true;
}
}
else
{
failed++;
errorBuffer.Add(outcome.Error);
}
if (recordBuffer.Count >= BatchSize)
{
FlushBuffers(recordBuffer, errorBuffer);
}
}
}
catch (OperationCanceledException)
{
cancelled = true;
}
finally
{
_cancellation = null;
FlushBuffers(recordBuffer, errorBuffer);
stopwatch.Stop();
}
var summary = BuildSummary(succeeded, failed, stopwatch.Elapsed, truncated, cancelled);
OnUi(() =>
{
StatusMessage = summary;
Progress = cancelled ? Progress : 1d;
});
_logger.LogInformation(
"Parsed with {Parser}: {Succeeded} record(s), {Failed} error(s) in {Elapsed}",
parser.Id,
succeeded,
failed,
stopwatch.Elapsed
);
}
///
/// Builds the outcome line out of translated fragments.
///
///
/// Assembled from plural-aware pieces rather than one format string per case: Russian needs
/// three forms for a counted noun, so "{0} records" with an English plural glued on cannot be
/// translated correctly.
///
internal static string BuildSummary(int succeeded, int failed, TimeSpan elapsed, bool truncated, bool cancelled)
{
var loc = Localizer.Instance;
var records = loc.Plural("Parse.Count.Records", succeeded);
var milliseconds = elapsed.TotalMilliseconds.ToString("N0", CultureInfo.CurrentCulture);
var text = loc.Format(cancelled ? "Parse.Status.Cancelled" : "Parse.Status.Done", records, milliseconds);
if (failed > 0)
{
text = loc.Format("Parse.Status.WithErrors", text.TrimEnd('.'), loc.Plural("Parse.Count.Errors", failed));
}
if (truncated)
{
text += " " + loc.Format("Parse.Status.Truncated", loc.Plural("Parse.Count.Records", MaxDisplayedRecords));
}
return text;
}
private void FlushBuffers(List records, List errors)
{
if (records.Count == 0 && errors.Count == 0)
{
return;
}
// Copy before clearing: the scheduled callback may run after the loop has refilled these.
var recordBatch = records.ToArray();
var errorBatch = errors.ToArray();
records.Clear();
errors.Clear();
OnUi(() =>
{
foreach (var record in recordBatch)
{
Records.Add(record);
}
foreach (var error in errorBatch)
{
Errors.Add(new ParseErrorViewModel(error));
}
});
}
///
/// The pool is a singleton and would otherwise keep this page alive for the process.
public void Dispose()
{
_proxyPool.Changed -= OnProxyPoolChanged;
_proxyChanged.Dispose();
GC.SuppressFinalize(this);
}
///
protected override void OnLanguageChanged()
{
base.OnLanguageChanged();
this.RaisePropertyChanged(nameof(ProxyRequiredMessage));
// The summary and the listed errors were both rendered in the previous language.
StatusMessage = null;
foreach (var error in Errors)
{
error.Refresh();
}
}
private void ClearResults()
{
Records.Clear();
Errors.Clear();
}
private void OnCommandFailed(Exception exception)
{
_logger.LogError(exception, "Parse failed");
OnUi(() => StatusMessage = Localizer.Instance.Format("Parse.Status.Failed", exception.Message));
}
/// Marshals a mutation onto the UI thread; the parse loop runs on the thread pool.
private void OnUi(Action action) => _mainThread.Schedule(action);
private static string SampleFor(string parserId) =>
parserId switch
{
"key-value" => """
# Sample configuration
host = localhost
port: 8080
enabled = true
name = av-parser
""",
_ => """
id,name,role
1,Ada Lovelace,Analyst
2,Grace Hopper,Compiler
3,Alan Turing,Cryptanalyst
""",
};
private static string LargeSampleFor(string parserId)
{
const int rows = 50_000;
var text = new StringBuilder(rows * 24);
if (parserId == "key-value")
{
for (var i = 0; i < rows; i++)
{
text.Append("key").Append(i).Append(" = value").Append(i).Append('\n');
}
return text.ToString();
}
text.Append("id,name,score\n");
for (var i = 0; i < rows; i++)
{
text.Append(i).Append(",item-").Append(i).Append(',').Append(i % 100).Append('\n');
}
return text.ToString();
}
}