rogee/wx-win-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Chunk connection-authorized sync batches

Browse Source
This commit is contained in:
rogee
2026-09-22 17:24:02 +08:00
parent 9121cd44a7
commit e057c5cdb3
1 changed files with 30 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-agent/WxAgent.Service/RemoteAgentHostedService.cs
+26 -4
View File
@@ -192,14 +192,36 @@ public sealed class RemoteAgentHostedService(
continue;
var sequence = sourceChanged ? 1 : checkpoint.ConfirmedSequence + 1;
var remainingConversations = collection.Conversations.ToList();
var remainingMessages = collection.Messages.ToList();
var cursorStartJson = SerializeCursors(cursorStart);
var cursorEndJson = SerializeCursors(collection.NextCursors);
var allQueued = true;
RemoteSyncBatch? lastQueuedBatch = null;
while (remainingConversations.Count > 0 || remainingMessages.Count > 0 || lastQueuedBatch is null)
{
var conversationCount = Math.Min(remainingConversations.Count, RemoteDataProtocol.MaxBatchMessages);
var conversations = remainingConversations.Take(conversationCount).ToArray();
remainingConversations.RemoveRange(0, conversationCount);
var messageCount = Math.Min(remainingMessages.Count, RemoteDataProtocol.MaxBatchMessages - conversations.Length);
var messages = remainingMessages.Take(messageCount).ToArray();
remainingMessages.RemoveRange(0, messageCount);
var batch = new RemoteSyncBatch(
remote.NodeId!, account.AccountId, Guid.NewGuid().ToString("N"), collection.SourceGeneration,
"messages", sequence, SerializeCursors(cursorStart), SerializeCursors(collection.NextCursors),
string.Empty, collection.IsComplete ? "complete" : "partial", collection.Conversations, collection.Messages);
"messages", sequence++, cursorStartJson, cursorEndJson, string.Empty,
collection.IsComplete ? "complete" : "partial", conversations, messages);
batch = batch with { PayloadHash = RemoteDataBatchAuthorization.ComputePayloadHash(batch) };
var queued = queue.Enqueue(reporting, batch);
if (queued.Accepted && queued.Batch is not null)
state.MarkCollected(queued.Batch, collection.UnavailableSources);
if (!queued.Accepted || queued.Batch is null)
{
allQueued = false;
break;
}
lastQueuedBatch = queued.Batch;
cursorStartJson = cursorEndJson;
}
if (allQueued && lastQueuedBatch is not null)
state.MarkCollected(lastQueuedBatch, collection.UnavailableSources);
if (!collection.IsComplete)
logger.LogWarning("Data sync is partial; accountId={AccountId}; unavailableSources={UnavailableSourceCount}.", account.AccountId, collection.UnavailableSources.Count);
}