更新项目配置，新增设备同步模块，优化WebSocket和Swagger配置，增强SCADA系统的免登录接口，支持数据字典项和登录日志的免登录查询与记录。调整Java编译设置，确保更好的开发体验。
This commit is contained in:
geht
@@ -0,0 +1,233 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Prism.Events;
|
||||
using System.IO;
|
||||
using System.Net.WebSockets;
|
||||
using System.Text;
|
||||
using YY.Admin.Core.Events;
|
||||
using YY.Admin.Core.Services;
|
||||
using YY.Admin.Infrastructure.Storage;
|
||||
|
||||
namespace YY.Admin.Infrastructure.Hubs;
|
||||
|
||||
public class StompWebSocketService : ISignalRService
|
||||
{
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IEventAggregator _eventAggregator;
|
||||
private readonly TokenStore _tokenStore;
|
||||
private ClientWebSocket? _socket;
|
||||
private string _deviceId = "default-device";
|
||||
private string _token = string.Empty;
|
||||
|
||||
public StompWebSocketService(
|
||||
IConfiguration configuration,
|
||||
IEventAggregator eventAggregator,
|
||||
TokenStore tokenStore)
|
||||
{
|
||||
_configuration = configuration;
|
||||
_eventAggregator = eventAggregator;
|
||||
_tokenStore = tokenStore;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task ConnectAsync(string token, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_token = token ?? string.Empty;
|
||||
await ConnectUnifiedDeviceChannelAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task ConnectUnifiedDeviceChannelAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var anonymous = _configuration.GetValue("JeecgIntegration:AnonymousMode", true);
|
||||
if (anonymous)
|
||||
{
|
||||
_token = string.Empty;
|
||||
}
|
||||
else if (string.IsNullOrWhiteSpace(_token))
|
||||
{
|
||||
_token = await _tokenStore.GetTokenAsync(cancellationToken).ConfigureAwait(false) ?? string.Empty;
|
||||
}
|
||||
|
||||
_deviceId = ResolveDeviceId(_token);
|
||||
var wsUrl = ResolveWsUrl();
|
||||
var retryDelays = new[] { 0, 2, 5, 10, 30 };
|
||||
|
||||
foreach (var delay in retryDelays)
|
||||
{
|
||||
if (delay > 0)
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(delay), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_socket?.Dispose();
|
||||
_socket = new ClientWebSocket();
|
||||
_socket.Options.AddSubProtocol("v12.stomp");
|
||||
await _socket.ConnectAsync(new Uri(wsUrl), cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var connectFrame = anonymous || string.IsNullOrWhiteSpace(_token)
|
||||
? "CONNECT\naccept-version:1.2\nheart-beat:10000,10000\n\n\0"
|
||||
: BuildConnectFrame(_token);
|
||||
await SendFrameAsync(connectFrame, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
// 用户镜像变更：与后端 /topic/sync/jeecg-users 对齐（设备同步统一线路）
|
||||
await SendFrameAsync(BuildSubscribeFrame("sub-jeecg-users", "/topic/sync/jeecg-users"), cancellationToken).ConfigureAwait(false);
|
||||
|
||||
// 非免密时同时订阅设备点对点指令队列
|
||||
if (!anonymous && !string.IsNullOrWhiteSpace(_token))
|
||||
{
|
||||
await SendFrameAsync(BuildSubscribeFrame("sub-device-command", $"/user/{_deviceId}/queue/command"), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_eventAggregator.GetEvent<NetworkStatusChangedEvent>().Publish(new NetworkStatusChangedPayload
|
||||
{
|
||||
IsOnline = true,
|
||||
ChangedAt = DateTime.UtcNow
|
||||
});
|
||||
|
||||
_ = Task.Run(() => ReceiveLoopAsync(cancellationToken), cancellationToken);
|
||||
return;
|
||||
}
|
||||
catch
|
||||
{
|
||||
_eventAggregator.GetEvent<NetworkStatusChangedEvent>().Publish(new NetworkStatusChangedPayload
|
||||
{
|
||||
IsOnline = false,
|
||||
ChangedAt = DateTime.UtcNow
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task SendDeviceStatusAsync(object status, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_socket == null || _socket.State != WebSocketState.Open)
|
||||
{
|
||||
return;
|
||||
}
|
||||
var json = System.Text.Json.JsonSerializer.Serialize(status);
|
||||
var frame = $"SEND\n" +
|
||||
$"destination:/app/device/status\n" +
|
||||
$"content-type:application/json\n" +
|
||||
$"content-length:{Encoding.UTF8.GetByteCount(json)}\n\n" +
|
||||
$"{json}\0";
|
||||
await SendFrameAsync(frame, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task ReceiveLoopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_socket == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
var buffer = new byte[8192];
|
||||
while (_socket.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
using var ms = new MemoryStream();
|
||||
WebSocketReceiveResult result;
|
||||
do
|
||||
{
|
||||
result = await _socket.ReceiveAsync(new ArraySegment<byte>(buffer), cancellationToken).ConfigureAwait(false);
|
||||
if (result.MessageType == WebSocketMessageType.Close)
|
||||
{
|
||||
_eventAggregator.GetEvent<NetworkStatusChangedEvent>().Publish(new NetworkStatusChangedPayload
|
||||
{
|
||||
IsOnline = false,
|
||||
ChangedAt = DateTime.UtcNow
|
||||
});
|
||||
await ConnectUnifiedDeviceChannelAsync(cancellationToken).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
ms.Write(buffer, 0, result.Count);
|
||||
} while (!result.EndOfMessage);
|
||||
|
||||
var text = Encoding.UTF8.GetString(ms.ToArray());
|
||||
if (!text.StartsWith("MESSAGE", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
var idx = text.IndexOf("\n\n", StringComparison.Ordinal);
|
||||
if (idx < 0)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
var body = text[(idx + 2)..].TrimEnd('\0');
|
||||
_eventAggregator.GetEvent<RemoteCommandReceivedEvent>().Publish(new RemoteCommandPayload
|
||||
{
|
||||
DeviceId = _deviceId,
|
||||
CommandJson = body
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SendFrameAsync(string frame, CancellationToken cancellationToken)
|
||||
{
|
||||
if (_socket == null || _socket.State != WebSocketState.Open)
|
||||
{
|
||||
return;
|
||||
}
|
||||
var data = Encoding.UTF8.GetBytes(frame);
|
||||
await _socket.SendAsync(new ArraySegment<byte>(data), WebSocketMessageType.Text, true, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private string ResolveWsUrl()
|
||||
{
|
||||
var baseUrl = _configuration.GetValue<string>("JeecgIntegration:BaseUrl")?.TrimEnd('/');
|
||||
if (string.IsNullOrWhiteSpace(baseUrl))
|
||||
{
|
||||
return "ws://127.0.0.1:8080/jeecg-boot/ws/device/websocket";
|
||||
}
|
||||
if (baseUrl.StartsWith("https://", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return "wss://" + baseUrl["https://".Length..] + "/ws/device/websocket";
|
||||
}
|
||||
return "ws://" + baseUrl["http://".Length..] + "/ws/device/websocket";
|
||||
}
|
||||
|
||||
private static string ResolveDeviceId(string token)
|
||||
{
|
||||
try
|
||||
{
|
||||
var parts = token.Split('.');
|
||||
if (parts.Length < 2)
|
||||
{
|
||||
return "default-device";
|
||||
}
|
||||
var payload = parts[1].Replace('-', '+').Replace('_', '/');
|
||||
payload = payload.PadRight(payload.Length + (4 - payload.Length % 4) % 4, '=');
|
||||
var json = Encoding.UTF8.GetString(Convert.FromBase64String(payload));
|
||||
using var doc = System.Text.Json.JsonDocument.Parse(json);
|
||||
if (doc.RootElement.TryGetProperty("deviceId", out var deviceId))
|
||||
{
|
||||
return deviceId.GetString() ?? "default-device";
|
||||
}
|
||||
if (doc.RootElement.TryGetProperty("username", out var username))
|
||||
{
|
||||
return username.GetString() ?? "default-device";
|
||||
}
|
||||
return "default-device";
|
||||
}
|
||||
catch
|
||||
{
|
||||
return "default-device";
|
||||
}
|
||||
}
|
||||
|
||||
private static string BuildConnectFrame(string token)
|
||||
{
|
||||
return "CONNECT\n" +
|
||||
"accept-version:1.2\n" +
|
||||
"heart-beat:10000,10000\n" +
|
||||
$"Authorization:Bearer {token}\n\n\0";
|
||||
}
|
||||
|
||||
private static string BuildSubscribeFrame(string subscriptionId, string destination)
|
||||
{
|
||||
return "SUBSCRIBE\n" +
|
||||
$"id:{subscriptionId}\n" +
|
||||
$"destination:{destination}\n" +
|
||||
"ack:auto\n\n\0";
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Prism.Events;
|
||||
using System.Net.Http;
|
||||
using YY.Admin.Core.Events;
|
||||
using YY.Admin.Core.Services;
|
||||
|
||||
namespace YY.Admin.Infrastructure.Network;
|
||||
|
||||
public class NetworkMonitor : INetworkMonitor
|
||||
{
|
||||
private readonly IHttpClientFactory _httpClientFactory;
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IEventAggregator _eventAggregator;
|
||||
private readonly SemaphoreSlim _startLock = new(1, 1);
|
||||
private Task? _loopTask;
|
||||
private CancellationTokenSource? _cts;
|
||||
private volatile bool _isOnline;
|
||||
|
||||
public NetworkMonitor(
|
||||
IHttpClientFactory httpClientFactory,
|
||||
IConfiguration configuration,
|
||||
IEventAggregator eventAggregator)
|
||||
{
|
||||
_httpClientFactory = httpClientFactory;
|
||||
_configuration = configuration;
|
||||
_eventAggregator = eventAggregator;
|
||||
}
|
||||
|
||||
public bool IsOnline => _isOnline;
|
||||
|
||||
public event Action<bool>? StatusChanged;
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
await _startLock.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
if (_cts != null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
_loopTask = Task.Run(() => MonitorLoopAsync(_cts.Token), _cts.Token);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_startLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task MonitorLoopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(10));
|
||||
while (await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
var online = await ProbeAsync(cancellationToken).ConfigureAwait(false);
|
||||
if (online == _isOnline)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
_isOnline = online;
|
||||
StatusChanged?.Invoke(online);
|
||||
_eventAggregator.GetEvent<NetworkStatusChangedEvent>().Publish(new NetworkStatusChangedPayload
|
||||
{
|
||||
IsOnline = online,
|
||||
ChangedAt = DateTime.UtcNow
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<bool> ProbeAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var baseUrl = _configuration.GetValue<string>("JeecgIntegration:BaseUrl")?.TrimEnd('/');
|
||||
if (string.IsNullOrWhiteSpace(baseUrl))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// 探活策略：优先调用免登录接口，失败后再降级到健康检查接口
|
||||
var probeUrls = new[]
|
||||
{
|
||||
$"{baseUrl}/sys/user/scada/queryUser?current=1&pageSize=1",
|
||||
$"{baseUrl}/sys/dict/scada/queryDictItem?pageNo=1&pageSize=1",
|
||||
$"{baseUrl}/actuator/health"
|
||||
};
|
||||
|
||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeoutCts.CancelAfter(TimeSpan.FromSeconds(3));
|
||||
var client = _httpClientFactory.CreateClient("JeecgApi");
|
||||
foreach (var url in probeUrls)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var req = new HttpRequestMessage(HttpMethod.Get, url);
|
||||
using var resp = await client.SendAsync(req, timeoutCts.Token).ConfigureAwait(false);
|
||||
if (resp.IsSuccessStatusCode)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
catch
|
||||
{
|
||||
// 当前探活地址失败后继续尝试下一个降级地址
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
using SqlSugar;
|
||||
|
||||
namespace YY.Admin.Infrastructure.Storage;
|
||||
|
||||
public class TokenStore
|
||||
{
|
||||
private const string TokenKey = "DeviceToken";
|
||||
private readonly ISqlSugarClient _db;
|
||||
|
||||
public TokenStore(ISqlSugarClient db)
|
||||
{
|
||||
_db = db;
|
||||
}
|
||||
|
||||
public async Task UpdateTokenAsync(string token, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var value = token ?? string.Empty;
|
||||
await EnsureTableAsync(cancellationToken).ConfigureAwait(false);
|
||||
var sql = "INSERT INTO app_config(config_key,config_value,updated_at) VALUES(@key,@val,@ts) " +
|
||||
"ON CONFLICT(config_key) DO UPDATE SET config_value=@val, updated_at=@ts;";
|
||||
var now = DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss");
|
||||
await _db.Ado.ExecuteCommandAsync(sql, new[]
|
||||
{
|
||||
new SugarParameter("@key", TokenKey),
|
||||
new SugarParameter("@val", value),
|
||||
new SugarParameter("@ts", now)
|
||||
}).ConfigureAwait(false);
|
||||
_ = cancellationToken;
|
||||
}
|
||||
|
||||
public async Task<string> GetTokenAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
await EnsureTableAsync(cancellationToken).ConfigureAwait(false);
|
||||
var sql = "SELECT config_value FROM app_config WHERE config_key=@key LIMIT 1;";
|
||||
var token = await _db.Ado.GetStringAsync(sql, new[] { new SugarParameter("@key", TokenKey) }).ConfigureAwait(false);
|
||||
return token ?? string.Empty;
|
||||
}
|
||||
|
||||
private async Task EnsureTableAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
const string sql = "CREATE TABLE IF NOT EXISTS app_config(" +
|
||||
"config_key TEXT PRIMARY KEY," +
|
||||
"config_value TEXT NULL," +
|
||||
"updated_at TEXT NULL);";
|
||||
await _db.Ado.ExecuteCommandAsync(sql).ConfigureAwait(false);
|
||||
_ = cancellationToken;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
using System.Net.Http.Json;
|
||||
using System.Net.Http;
|
||||
using YY.Admin.Core.Models;
|
||||
using YY.Admin.Infrastructure.Storage;
|
||||
|
||||
namespace YY.Admin.Infrastructure.Sync;
|
||||
|
||||
public class HttpSyncClient
|
||||
{
|
||||
private readonly IHttpClientFactory _httpClientFactory;
|
||||
private readonly TokenStore _tokenStore;
|
||||
|
||||
private sealed class SyncBatchItem
|
||||
{
|
||||
public string MessageId { get; set; } = string.Empty;
|
||||
public string AggregateType { get; set; } = string.Empty;
|
||||
public string AggregateId { get; set; } = string.Empty;
|
||||
public string EventType { get; set; } = string.Empty;
|
||||
public string Payload { get; set; } = string.Empty;
|
||||
public DateTime OccurredAt { get; set; }
|
||||
}
|
||||
|
||||
public HttpSyncClient(IHttpClientFactory httpClientFactory, TokenStore tokenStore)
|
||||
{
|
||||
_httpClientFactory = httpClientFactory;
|
||||
_tokenStore = tokenStore;
|
||||
}
|
||||
|
||||
public async Task<bool> SendBatchAsync(IReadOnlyCollection<OutboxMessage> messages, CancellationToken cancellationToken)
|
||||
{
|
||||
if (messages.Count == 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
var client = _httpClientFactory.CreateClient("JeecgApi");
|
||||
var body = messages.Select(m => new SyncBatchItem
|
||||
{
|
||||
MessageId = m.Id,
|
||||
AggregateType = m.AggregateType,
|
||||
AggregateId = m.AggregateId,
|
||||
EventType = m.EventType,
|
||||
Payload = m.Payload,
|
||||
OccurredAt = m.CreatedAt
|
||||
}).ToList();
|
||||
|
||||
using var response = await client.PostAsJsonAsync("/sys/sync/batch", body, cancellationToken).ConfigureAwait(false);
|
||||
if (response.Headers.TryGetValues("X-Refresh-Token", out var values))
|
||||
{
|
||||
var refreshed = values.FirstOrDefault();
|
||||
if (!string.IsNullOrWhiteSpace(refreshed))
|
||||
{
|
||||
await _tokenStore.UpdateTokenAsync(refreshed, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
return response.IsSuccessStatusCode;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
using YY.Admin.Core.Services;
|
||||
using YY.Admin.Core.Sync;
|
||||
|
||||
namespace YY.Admin.Infrastructure.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// 用户镜像拉取入队：统一走 Outbox（断网续传），与设备同步模块同一条基础设施线路。
|
||||
/// </summary>
|
||||
public sealed class JeecgUserMirrorPullOutbox : IJeecgUserMirrorPullOutbox
|
||||
{
|
||||
private readonly OutboxProcessor _outboxProcessor;
|
||||
|
||||
public JeecgUserMirrorPullOutbox(OutboxProcessor outboxProcessor)
|
||||
{
|
||||
_outboxProcessor = outboxProcessor;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task EnqueuePullAsync(string eventType, string? payloadJson, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var payload = string.IsNullOrWhiteSpace(payloadJson) ? "{}" : payloadJson;
|
||||
return _outboxProcessor.EnqueueAsync(
|
||||
JeecgUserMirrorOutbox.AggregateType,
|
||||
"mirror",
|
||||
eventType,
|
||||
new { source = "unified-device-channel", detail = payload },
|
||||
cancellationToken);
|
||||
}
|
||||
}
|
||||
232
yy-admin-master/YY.Admin/Infrastructure/Sync/OutboxProcessor.cs
Normal file
232
yy-admin-master/YY.Admin/Infrastructure/Sync/OutboxProcessor.cs
Normal file
@@ -0,0 +1,232 @@
|
||||
using SqlSugar;
|
||||
using System.Threading.Channels;
|
||||
using Prism.Events;
|
||||
using YY.Admin.Core.Events;
|
||||
using YY.Admin.Core.Models;
|
||||
using YY.Admin.Core.Services;
|
||||
using YY.Admin.Core.Sync;
|
||||
|
||||
namespace YY.Admin.Infrastructure.Sync;
|
||||
|
||||
public class OutboxProcessor
|
||||
{
|
||||
private readonly INetworkMonitor _networkMonitor;
|
||||
private readonly HttpSyncClient _httpSyncClient;
|
||||
private readonly IJeecgUserMirrorPullHandler _mirrorPullHandler;
|
||||
private readonly ISqlSugarClient _db;
|
||||
private readonly IEventAggregator _eventAggregator;
|
||||
private readonly Channel<OutboxMessage> _channel = Channel.CreateBounded<OutboxMessage>(new BoundedChannelOptions(1000)
|
||||
{
|
||||
SingleReader = true,
|
||||
SingleWriter = false,
|
||||
FullMode = BoundedChannelFullMode.Wait
|
||||
});
|
||||
private readonly SemaphoreSlim _flushLock = new(1, 1);
|
||||
|
||||
public OutboxProcessor(
|
||||
INetworkMonitor networkMonitor,
|
||||
HttpSyncClient httpSyncClient,
|
||||
IJeecgUserMirrorPullHandler mirrorPullHandler,
|
||||
ISqlSugarClient db,
|
||||
IEventAggregator eventAggregator)
|
||||
{
|
||||
_networkMonitor = networkMonitor;
|
||||
_httpSyncClient = httpSyncClient;
|
||||
_mirrorPullHandler = mirrorPullHandler;
|
||||
_db = db.AsTenant().GetConnectionScope("Slave");
|
||||
_eventAggregator = eventAggregator;
|
||||
_networkMonitor.StatusChanged += OnNetworkStatusChanged;
|
||||
}
|
||||
|
||||
private static bool IsJeecgUserMirrorMessage(OutboxMessage m) =>
|
||||
string.Equals(m.AggregateType, JeecgUserMirrorOutbox.AggregateType, StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
public async Task EnqueueAsync<T>(
|
||||
string aggregateType,
|
||||
string aggregateId,
|
||||
string eventType,
|
||||
T payload,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var now = DateTime.UtcNow;
|
||||
var message = new OutboxMessage
|
||||
{
|
||||
AggregateType = aggregateType,
|
||||
AggregateId = aggregateId,
|
||||
EventType = eventType,
|
||||
Payload = System.Text.Json.JsonSerializer.Serialize(payload),
|
||||
Status = 0,
|
||||
RetryCount = 0,
|
||||
CreatedAt = now
|
||||
};
|
||||
await EnsureTableAsync(cancellationToken).ConfigureAwait(false);
|
||||
await _db.Insertable(message).ExecuteCommandAsync(cancellationToken).ConfigureAwait(false);
|
||||
if (_networkMonitor.IsOnline)
|
||||
{
|
||||
await _channel.Writer.WriteAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task StartConsumerAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await EnsureTableAsync(cancellationToken).ConfigureAwait(false);
|
||||
_ = Task.Run(() => ConsumeLoopAsync(cancellationToken), cancellationToken);
|
||||
}
|
||||
|
||||
public async Task FlushPendingAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (!await _flushLock.WaitAsync(0, cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
var pending = await _db.Queryable<OutboxMessage>()
|
||||
.Where(x => x.Status == 0 && x.RetryCount < 5)
|
||||
.OrderBy(x => x.CreatedAt)
|
||||
.ToListAsync(cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
if (pending.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var mirror = pending.Where(IsJeecgUserMirrorMessage).ToList();
|
||||
var serverBatch = pending.Where(m => !IsJeecgUserMirrorMessage(m)).ToList();
|
||||
|
||||
foreach (var item in mirror)
|
||||
{
|
||||
var ok = await _mirrorPullHandler.ExecutePullAsync(cancellationToken).ConfigureAwait(false);
|
||||
if (ok)
|
||||
{
|
||||
await MarkSentAsync(item, cancellationToken).ConfigureAwait(false);
|
||||
_eventAggregator.GetEvent<SyncCompletedEvent>().Publish(item.AggregateId);
|
||||
}
|
||||
else
|
||||
{
|
||||
await MarkFailedAsync(item, "Jeecg用户镜像拉取失败", cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
if (serverBatch.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var success = await _httpSyncClient.SendBatchAsync(serverBatch, cancellationToken).ConfigureAwait(false);
|
||||
if (success)
|
||||
{
|
||||
var ids = serverBatch.Select(x => x.Id).ToArray();
|
||||
await _db.Updateable<OutboxMessage>()
|
||||
.SetColumns(x => new OutboxMessage
|
||||
{
|
||||
Status = 1,
|
||||
SentAt = DateTime.UtcNow,
|
||||
LastTriedAt = DateTime.UtcNow,
|
||||
ErrorMessage = null
|
||||
})
|
||||
.Where(x => ids.Contains(x.Id))
|
||||
.ExecuteCommandAsync(cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
foreach (var item in serverBatch)
|
||||
{
|
||||
_eventAggregator.GetEvent<SyncCompletedEvent>().Publish(item.AggregateId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var item in serverBatch)
|
||||
{
|
||||
await MarkFailedAsync(item, "批量同步失败", cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_flushLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ConsumeLoopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
while (await _channel.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
while (_channel.Reader.TryRead(out var message))
|
||||
{
|
||||
var success = IsJeecgUserMirrorMessage(message)
|
||||
? await _mirrorPullHandler.ExecutePullAsync(cancellationToken).ConfigureAwait(false)
|
||||
: await _httpSyncClient.SendBatchAsync(new[] { message }, cancellationToken).ConfigureAwait(false);
|
||||
if (success)
|
||||
{
|
||||
await MarkSentAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
_eventAggregator.GetEvent<SyncCompletedEvent>().Publish(message.AggregateId);
|
||||
}
|
||||
else
|
||||
{
|
||||
await MarkFailedAsync(message, "实时同步失败", cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task MarkSentAsync(OutboxMessage message, CancellationToken cancellationToken)
|
||||
{
|
||||
await _db.Updateable<OutboxMessage>()
|
||||
.SetColumns(x => new OutboxMessage
|
||||
{
|
||||
Status = 1,
|
||||
SentAt = DateTime.UtcNow,
|
||||
LastTriedAt = DateTime.UtcNow,
|
||||
ErrorMessage = null
|
||||
})
|
||||
.Where(x => x.Id == message.Id)
|
||||
.ExecuteCommandAsync(cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task MarkFailedAsync(OutboxMessage message, string error, CancellationToken cancellationToken)
|
||||
{
|
||||
var nextRetry = message.RetryCount + 1;
|
||||
var backoff = (int)Math.Pow(2, Math.Min(nextRetry, 5));
|
||||
await _db.Updateable<OutboxMessage>()
|
||||
.SetColumns(x => new OutboxMessage
|
||||
{
|
||||
RetryCount = nextRetry,
|
||||
Status = nextRetry >= 5 ? 2 : 0,
|
||||
ErrorMessage = error,
|
||||
LastTriedAt = DateTime.UtcNow
|
||||
})
|
||||
.Where(x => x.Id == message.Id)
|
||||
.ExecuteCommandAsync(cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
if (nextRetry < 5)
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(backoff), cancellationToken).ConfigureAwait(false);
|
||||
if (_networkMonitor.IsOnline)
|
||||
{
|
||||
var retryMessage = await _db.Queryable<OutboxMessage>()
|
||||
.FirstAsync(x => x.Id == message.Id, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
if (retryMessage != null && retryMessage.Status == 0)
|
||||
{
|
||||
await _channel.Writer.WriteAsync(retryMessage, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task EnsureTableAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_ = cancellationToken;
|
||||
await Task.Run(() => _db.CodeFirst.InitTables<OutboxMessage>(), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private void OnNetworkStatusChanged(bool isOnline)
|
||||
{
|
||||
if (!isOnline)
|
||||
{
|
||||
return;
|
||||
}
|
||||
_ = FlushPendingAsync(default);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user