refactor(agent): remove obsolete compatibility paths
This commit is contained in:
@@ -1,4 +1,3 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using WxAgent.Core;
|
||||
|
||||
@@ -20,11 +19,10 @@ public sealed class EventPump(IAgentBackend backend, EventHub hub, ServiceOption
|
||||
{
|
||||
await foreach (var item in source.ListenAsync(stoppingToken))
|
||||
{
|
||||
var runtime = await LoadRuntimeConfigurationAsync(options, stoppingToken);
|
||||
if (remoteQueue is not null && runtime.Remote.IsConfigured && item.Kind == "message"
|
||||
if (remoteQueue is not null && options.Remote is { IsConfigured: true } remote && item.Kind == "message"
|
||||
&& item.ChatId is { Length: > 0 } chatId && item.ChatType is { } chatType)
|
||||
{
|
||||
_ = remoteQueue.Enqueue(runtime.Reporting, runtime.Remote.NodeId!, item.AccountId, chatId, chatType,
|
||||
_ = remoteQueue.Enqueue(options.Reporting, remote.NodeId!, item.AccountId, chatId, chatType,
|
||||
item.Kind, item.At, item.Content);
|
||||
}
|
||||
var localItem = item with { Content = null };
|
||||
@@ -37,18 +35,4 @@ public sealed class EventPump(IAgentBackend backend, EventHub hub, ServiceOption
|
||||
catch (Exception) { await Task.Delay(TimeSpan.FromSeconds(2), stoppingToken); }
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task<RemoteNodeConfiguration> LoadRuntimeConfigurationAsync(ServiceOptions options, CancellationToken cancellationToken)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
||||
return new RemoteNodeConfiguration { Remote = options.Remote ?? new RemoteAgentOptions(), Reporting = options.Reporting };
|
||||
try
|
||||
{
|
||||
return await RemoteNodeConfigurationStore.LoadAsync(options.RemoteConfigurationFile, cancellationToken);
|
||||
}
|
||||
catch (Exception exception) when (exception is IOException or JsonException or WxAgentException)
|
||||
{
|
||||
return new RemoteNodeConfiguration();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,31 +19,17 @@ public sealed class RemoteAgentHostedService(
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
RemoteAgentOptions? configuredRemote = options.Remote;
|
||||
if (!string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
||||
{
|
||||
try
|
||||
{
|
||||
configuredRemote = (await RemoteNodeConfigurationStore.LoadAsync(options.RemoteConfigurationFile, stoppingToken)).Remote;
|
||||
}
|
||||
catch (Exception exception) when (exception is IOException or JsonException or WxAgentException)
|
||||
{
|
||||
logger.LogWarning("Remote agent is disabled because the external configuration could not be loaded; type={ExceptionType}.", exception.GetType().Name);
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (configuredRemote is null || !configuredRemote.IsConfigured)
|
||||
var remote = options.Remote;
|
||||
if (remote is null || !remote.IsConfigured)
|
||||
{
|
||||
logger.LogInformation("Remote agent is disabled because authAddress, token or nodeId is not configured.");
|
||||
return;
|
||||
}
|
||||
var remote = configuredRemote;
|
||||
|
||||
using var client = new RemoteControlClient(remote);
|
||||
var ledger = new RemoteTaskLedger(Path.Combine(options.DataDirectory, "remote-task-ledger.json"));
|
||||
var eventQueue = remoteQueue ?? new RemoteEventQueue(Path.Combine(options.DataDirectory, "remote-event-queue.json"));
|
||||
// A configured Agent connection is the data-sync authorization boundary.
|
||||
// EnableDataSync remains only as a read-compatibility property for old service.json files.
|
||||
var dataQueue = new RemoteDataBatchQueue(
|
||||
Path.Combine(options.DataDirectory, "remote-data-queue.json"), options.DataSyncQueueMaxItems, options.DataSyncQueueMaxBytes);
|
||||
var syncState = new RemoteDataSyncStateStore(Path.Combine(options.DataDirectory, "remote-data-sync-state.json"));
|
||||
@@ -52,33 +38,14 @@ public sealed class RemoteAgentHostedService(
|
||||
string? registeredActiveAccountId = null;
|
||||
bool? registeredActiveAccountVerified = null;
|
||||
long registeredReportingConfigVersion = -1;
|
||||
var initialReporting = options.Reporting;
|
||||
if (!string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
||||
{
|
||||
try
|
||||
{
|
||||
initialReporting = (await LoadRuntimeConfigurationAsync(options, stoppingToken)).Reporting;
|
||||
}
|
||||
catch (Exception exception) when (exception is IOException or JsonException or WxAgentException)
|
||||
{
|
||||
logger.LogWarning("Remote reporting configuration could not be loaded for offline collection; type={ExceptionType}.", exception.GetType().Name);
|
||||
}
|
||||
}
|
||||
ReportingConfig? activeReporting = ConnectionAuthorizedReporting(initialReporting, null);
|
||||
ReportingConfig? activeReporting = ConnectionAuthorizedReporting(options.Reporting, null);
|
||||
var retry = RetryDelay;
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
var runtime = await LoadRuntimeConfigurationAsync(options, stoppingToken);
|
||||
if (!runtime.Remote.IsConfigured || !runtime.Remote.Equals(remote))
|
||||
{
|
||||
logger.LogWarning("Remote configuration changed or is not valid; remote work is paused until the service is restarted.");
|
||||
await DelayAsync(RetryDelay, stoppingToken);
|
||||
continue;
|
||||
}
|
||||
var snapshot = await ReadSnapshotAsync(remote, stoppingToken);
|
||||
var reporting = ConnectionAuthorizedReporting(runtime.Reporting, snapshot);
|
||||
var reporting = ConnectionAuthorizedReporting(options.Reporting, snapshot);
|
||||
activeReporting = reporting;
|
||||
var registrationNeedsRefresh = client.AuthState != RemoteAuthState.Authenticated
|
||||
|| !string.Equals(registeredActiveAccountId, snapshot.ActiveAccountId, StringComparison.Ordinal)
|
||||
@@ -293,11 +260,6 @@ public sealed class RemoteAgentHostedService(
|
||||
{
|
||||
remote = await client.GetDataSyncStatusAsync(account.AccountId, cancellationToken, streamKey).ConfigureAwait(false);
|
||||
}
|
||||
catch (RemoteClientException exception) when (exception.StatusCode == 404 && exception.Code == "NotFound")
|
||||
{
|
||||
// Older control planes do not expose the reconciliation endpoint; retain the legacy ACK path.
|
||||
continue;
|
||||
}
|
||||
catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized")
|
||||
{
|
||||
var droppedOnRevoke = queue.DropAllForAccount(account.AccountId);
|
||||
@@ -879,13 +841,6 @@ public sealed class RemoteAgentHostedService(
|
||||
JsonElement Content,
|
||||
IReadOnlyList<RemoteReportingScope> ChatScopes);
|
||||
|
||||
private static async Task<RemoteNodeConfiguration> LoadRuntimeConfigurationAsync(ServiceOptions options, CancellationToken cancellationToken)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
||||
return new RemoteNodeConfiguration { Remote = options.Remote ?? new RemoteAgentOptions(), Reporting = options.Reporting };
|
||||
return await RemoteNodeConfigurationStore.LoadAsync(options.RemoteConfigurationFile, cancellationToken);
|
||||
}
|
||||
|
||||
private static async Task ReplayUnreportedResultsAsync(
|
||||
RemoteControlClient client,
|
||||
RemoteTaskLedger ledger,
|
||||
|
||||
@@ -54,10 +54,10 @@ public static class ServiceHost
|
||||
builder.Services.AddSingleton<AccountBindingStore>();
|
||||
builder.Services.AddSingleton<OperationQueue>();
|
||||
builder.Services.AddHostedService(p => p.GetRequiredService<OperationQueue>());
|
||||
if (options.Remote is not null || !string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
||||
if (options.Remote is not null)
|
||||
builder.Services.AddSingleton(new RemoteEventQueue(Path.Combine(options.DataDirectory, "remote-event-queue.json")));
|
||||
builder.Services.AddHostedService<EventPump>();
|
||||
if (options.Remote is not null || !string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
||||
if (options.Remote is not null)
|
||||
builder.Services.AddHostedService<RemoteAgentHostedService>();
|
||||
builder.Services.AddHttpContextAccessor();
|
||||
builder.Services.AddScoped<AgentService>();
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
using System.Net;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using System.Text.Json.Serialization;
|
||||
using WxAgent.Core;
|
||||
|
||||
namespace WxAgent.Service;
|
||||
@@ -21,27 +20,16 @@ public sealed class ServiceOptions
|
||||
public required string DataDirectory { get; init; }
|
||||
public RemoteAgentOptions? Remote { get; init; }
|
||||
public ReportingConfig Reporting { get; init; } = new();
|
||||
public string? RemoteConfigurationFile { get; init; }
|
||||
// Connection authorization covers data synchronization; validation writes remain separately gated.
|
||||
public bool EnableValidationOperations { get; init; }
|
||||
public bool EnableListenerEvents { get; init; }
|
||||
public bool PreventAutoLock { get; init; }
|
||||
// Kept as a compatibility switch for older service.json files; a configured Agent connection authorizes sync.
|
||||
public bool EnableDataSync { get; init; } = true;
|
||||
public int DataSyncIntervalSeconds { get; init; } = 5;
|
||||
public int DataSyncBatchLimit { get; init; } = 100;
|
||||
public int DataSyncOverlapRows { get; init; } = 1;
|
||||
public int DataSyncQueueMaxItems { get; init; } = 1000;
|
||||
public long DataSyncQueueMaxBytes { get; init; } = 16 * 1024 * 1024;
|
||||
|
||||
// Kept only so older service.json files can be loaded and rewritten by the tray.
|
||||
[JsonIgnore]
|
||||
[Obsolete("Internal compatibility field; the value is ignored.")]
|
||||
public int QueueCapacity { get; init; } = 100;
|
||||
[JsonIgnore]
|
||||
[Obsolete("Internal compatibility field; the value is ignored.")]
|
||||
public string ListenerSession { get; init; } = "文件传输助手";
|
||||
|
||||
public void Validate()
|
||||
{
|
||||
if (!Uri.TryCreate(ListenUrl, UriKind.Absolute, out var uri) || uri.Scheme != "http" ||
|
||||
|
||||
Reference in New Issue
Block a user