481 lines
22 KiB
C#
481 lines
22 KiB
C#
using System.Buffers.Binary;
|
||
|
||
namespace ICRebuild.Core;
|
||
|
||
public static class ActiveSendScope
|
||
{
|
||
[ThreadStatic] private static int depth;
|
||
public static bool Active => depth > 0;
|
||
public static IDisposable Enter()
|
||
{
|
||
depth++;
|
||
return new Scope();
|
||
}
|
||
private sealed class Scope : IDisposable
|
||
{
|
||
private int disposed;
|
||
public void Dispose()
|
||
{
|
||
if (Interlocked.Exchange(ref disposed, 1) == 0 && depth > 0) depth--;
|
||
}
|
||
}
|
||
}
|
||
|
||
public enum SendTransactionState
|
||
{
|
||
Queued,
|
||
SentNative,
|
||
RejectedNative,
|
||
Confirmed,
|
||
TimedOut,
|
||
Canceled
|
||
}
|
||
|
||
public sealed record SendSession(bool Ready, uint Territory, nint LocalAddress, uint LocalEntityId)
|
||
{
|
||
public static readonly SendSession Empty = new(false, 0, 0, 0);
|
||
public bool IsValid => Ready && Territory != 0 && LocalAddress != 0 && LocalEntityId is not (0 or 0xE0000000);
|
||
public bool Matches(SendSession other) =>
|
||
IsValid && other.IsValid && Territory == other.Territory &&
|
||
LocalAddress == other.LocalAddress && LocalEntityId == other.LocalEntityId;
|
||
}
|
||
|
||
public sealed record EncodedGamePacket(
|
||
string Name,
|
||
int Opcode,
|
||
byte[] Payload,
|
||
byte[] NativeBuffer,
|
||
int? ExpectedResponseOpcode = null,
|
||
Func<ReadOnlySpan<byte>, bool>? ResponseMatcher = null);
|
||
|
||
public sealed record SendTransactionSnapshot(
|
||
long Id,
|
||
string PacketName,
|
||
int Opcode,
|
||
SendTransactionState State,
|
||
long CreatedMs,
|
||
long? SentMs,
|
||
long? CompletedMs,
|
||
byte? NativeResult,
|
||
long? ResponseObservationId,
|
||
long? RoundTripMs,
|
||
string Detail);
|
||
|
||
public static class NativeSendPacket
|
||
{
|
||
public const string AuditedClientVersion = PacketParser.AuditedClientVersion;
|
||
public const string BaselineClientVersion = PacketParser.BaselineClientVersion;
|
||
public static bool IsAuditedVersion(string? version) => PacketParser.IsAuditedVersion(version);
|
||
public const int NativePrefixSize = 0x20;
|
||
public const int IpcHeaderSize = NativePacketCapture.IpcHeaderSize;
|
||
|
||
// Native ZoneUp (0x1D718C0) reads: +0x00 uint Opcode, +0x08 ulong (payloadLength + 0x10), +0x20 byte[] Payload.
|
||
public static byte[] BuildBuffer(int opcode, ReadOnlySpan<byte> payload)
|
||
{
|
||
if (opcode is <= 0 or > ushort.MaxValue) throw new ArgumentOutOfRangeException(nameof(opcode), "Opcode 超出有效范围");
|
||
if (payload.Length > NativePacketCapture.MaxPayload) throw new ArgumentOutOfRangeException(nameof(payload), "载荷超出 64 KiB 上限");
|
||
var buffer = new byte[NativePrefixSize + payload.Length];
|
||
BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(0, 4), (uint)opcode);
|
||
BinaryPrimitives.WriteUInt64LittleEndian(buffer.AsSpan(8, 8), (ulong)(payload.Length + IpcHeaderSize));
|
||
payload.CopyTo(buffer.AsSpan(NativePrefixSize));
|
||
return buffer;
|
||
}
|
||
}
|
||
|
||
public static class AuditedPacketEncoder
|
||
{
|
||
public static EncodedGamePacket EncodeEventStart(OpcodeMap map, string version, ulong targetObjectId, uint eventId, uint param = 0)
|
||
{
|
||
RequireAuditedVersion(version);
|
||
if (targetObjectId == 0 || targetObjectId == 0xE0000000) throw new ArgumentException("目标对象 ID 无效", nameof(targetObjectId));
|
||
if (eventId == 0) throw new ArgumentException("事件 ID 不能为 0", nameof(eventId));
|
||
var opcode = RequireOpcode(map, version, PacketDirection.Up, "UP_EventStart");
|
||
var responseOpcode = map.ResolveOpcode(version, PacketDirection.Down, "DOWN_EventStart");
|
||
var payload = new byte[0x10];
|
||
BinaryPrimitives.WriteUInt64LittleEndian(payload.AsSpan(0, 8), targetObjectId);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(8, 4), eventId);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(12, 4), param);
|
||
return new("UP_EventStart", opcode, payload, NativeSendPacket.BuildBuffer(opcode, payload), responseOpcode,
|
||
responseOpcode.HasValue ? p => p.Length >= 12 && BinaryPrimitives.ReadUInt32LittleEndian(p.Slice(8, 4)) == eventId : null);
|
||
}
|
||
|
||
public static EncodedGamePacket EncodeEventFinish(OpcodeMap map, string version, uint eventId, ushort sceneId, byte state, ReadOnlySpan<uint> args)
|
||
{
|
||
RequireAuditedVersion(version);
|
||
if (eventId == 0) throw new ArgumentException("事件 ID 不能为 0", nameof(eventId));
|
||
var (name, capacity, length) = SelectEventTier("UP_EventFinish", args.Length);
|
||
var opcode = RequireOpcode(map, version, PacketDirection.Up, name);
|
||
var responseOpcode = map.ResolveOpcode(version, PacketDirection.Down, "DOWN_EventFinish");
|
||
var payload = WriteEventPayload(eventId, sceneId, state, args, capacity, length);
|
||
return new(name, opcode, payload, NativeSendPacket.BuildBuffer(opcode, payload), responseOpcode,
|
||
responseOpcode.HasValue ? p => p.Length >= 4 && BinaryPrimitives.ReadUInt32LittleEndian(p[..4]) == eventId : null);
|
||
}
|
||
|
||
public static EncodedGamePacket EncodeEventAction(OpcodeMap map, string version, uint eventId, ushort sceneId, byte state, ReadOnlySpan<uint> args)
|
||
{
|
||
RequireAuditedVersion(version);
|
||
if (eventId == 0) throw new ArgumentException("事件 ID 不能为 0", nameof(eventId));
|
||
var (name, capacity, length) = SelectEventTier("UP_EventAction", args.Length);
|
||
var opcode = RequireOpcode(map, version, PacketDirection.Up, name);
|
||
var payload = WriteEventPayload(eventId, sceneId, state, args, capacity, length);
|
||
return new(name, opcode, payload, NativeSendPacket.BuildBuffer(opcode, payload));
|
||
}
|
||
|
||
public static EncodedGamePacket EncodeClientTrigger(OpcodeMap map, string version, uint commandId,
|
||
uint arg0 = 0, uint arg1 = 0, uint arg2 = 0, uint arg3 = 0, ulong targetId = 0)
|
||
{
|
||
RequireAuditedVersion(version);
|
||
if (commandId == 0) throw new ArgumentException("触发指令 ID 不能为 0", nameof(commandId));
|
||
var opcode = RequireOpcode(map, version, PacketDirection.Up, "UP_ClientTrigger");
|
||
var payload = new byte[0x20];
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(0, 4), commandId);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(4, 4), arg0);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(8, 4), arg1);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(12, 4), arg2);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(16, 4), arg3);
|
||
BinaryPrimitives.WriteUInt64LittleEndian(payload.AsSpan(24, 8), targetId);
|
||
return new("UP_ClientTrigger", opcode, payload, NativeSendPacket.BuildBuffer(opcode, payload));
|
||
}
|
||
|
||
public static EncodedGamePacket EncodeActionSend(OpcodeMap map, string version, uint actionId, byte actionKind,
|
||
ushort requestId, ulong targetId, ushort facing = 0, ushort targetFacing = 0, uint arg = 0)
|
||
{
|
||
RequireAuditedVersion(version);
|
||
if (actionId == 0) throw new ArgumentException("技能 ID 不能为 0", nameof(actionId));
|
||
if (requestId == 0) throw new ArgumentException("请求序号 RequestId 必须大于 0", nameof(requestId));
|
||
if (targetId == 0 || targetId == 0xE0000000) throw new ArgumentException("目标对象 ID 无效", nameof(targetId));
|
||
var opcode = RequireOpcode(map, version, PacketDirection.Up, "UP_ActionSend");
|
||
var responseOpcode = map.ResolveOpcode(version, PacketDirection.Down, "DOWN_Effect");
|
||
var payload = new byte[0x20];
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(0, 4), actionId);
|
||
payload[5] = actionKind;
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(6, 2), requestId);
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(8, 2), facing);
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(10, 2), targetFacing);
|
||
BinaryPrimitives.WriteUInt64LittleEndian(payload.AsSpan(16, 8), targetId);
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(24, 4), arg);
|
||
return new("UP_ActionSend", opcode, payload, NativeSendPacket.BuildBuffer(opcode, payload), responseOpcode,
|
||
responseOpcode.HasValue
|
||
? p => p.Length == 0x7C && p[0x21] <= 1 && BinaryPrimitives.ReadUInt16LittleEndian(p.Slice(0x18, 2)) == requestId
|
||
: null);
|
||
}
|
||
|
||
public static EncodedGamePacket EncodeActionSendPos(OpcodeMap map, string version, uint actionId, byte actionKind,
|
||
ushort requestId, System.Numerics.Vector3 position, ushort facing = 0, ushort targetFacing = 0)
|
||
{
|
||
RequireAuditedVersion(version);
|
||
if (actionId == 0) throw new ArgumentException("技能 ID 不能为 0", nameof(actionId));
|
||
if (requestId == 0) throw new ArgumentException("请求序号 RequestId 必须大于 0", nameof(requestId));
|
||
if (!float.IsFinite(position.X) || !float.IsFinite(position.Y) || !float.IsFinite(position.Z))
|
||
throw new ArgumentException("地面目标坐标包含无效浮点数", nameof(position));
|
||
var opcode = RequireOpcode(map, version, PacketDirection.Up, "UP_ActionSendPos");
|
||
var responseOpcode = map.ResolveOpcode(version, PacketDirection.Down, "DOWN_Effect");
|
||
var payload = new byte[0x18];
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(0, 4), actionId);
|
||
payload[5] = actionKind;
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(6, 2), requestId);
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(8, 2), facing);
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(10, 2), targetFacing);
|
||
BinaryPrimitives.WriteSingleLittleEndian(payload.AsSpan(12, 4), position.X);
|
||
BinaryPrimitives.WriteSingleLittleEndian(payload.AsSpan(16, 4), position.Y);
|
||
BinaryPrimitives.WriteSingleLittleEndian(payload.AsSpan(20, 4), position.Z);
|
||
return new("UP_ActionSendPos", opcode, payload, NativeSendPacket.BuildBuffer(opcode, payload), responseOpcode,
|
||
responseOpcode.HasValue
|
||
? p => p.Length == 0x7C && BinaryPrimitives.ReadUInt16LittleEndian(p.Slice(0x18, 2)) == requestId
|
||
: null);
|
||
}
|
||
|
||
private static (string Name, int Capacity, int Length) SelectEventTier(string prefix, int argCount) => argCount switch
|
||
{
|
||
< 0 or > 255 => throw new ArgumentOutOfRangeException(nameof(argCount), "事件参数数量超出 0..255 范围"),
|
||
<= 2 => (prefix, 2, 0x10),
|
||
<= 4 => ($"{prefix}4", 4, 0x18),
|
||
<= 8 => ($"{prefix}8", 8, 0x28),
|
||
<= 16 => ($"{prefix}16", 16, 0x48),
|
||
<= 32 => ($"{prefix}32", 32, 0x88),
|
||
<= 64 => ($"{prefix}64", 64, 0x108),
|
||
<= 128 => ($"{prefix}128", 128, 0x208),
|
||
_ => ($"{prefix}255", 255, 0x408)
|
||
};
|
||
|
||
private static byte[] WriteEventPayload(uint eventId, ushort sceneId, byte state, ReadOnlySpan<uint> args, int capacity, int length)
|
||
{
|
||
if (args.Length > capacity) throw new ArgumentOutOfRangeException(nameof(args));
|
||
var payload = new byte[length];
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(0, 4), eventId);
|
||
BinaryPrimitives.WriteUInt16LittleEndian(payload.AsSpan(4, 2), sceneId);
|
||
payload[6] = state;
|
||
payload[7] = (byte)args.Length;
|
||
for (var i = 0; i < args.Length; i++)
|
||
BinaryPrimitives.WriteUInt32LittleEndian(payload.AsSpan(8 + i * 4, 4), args[i]);
|
||
return payload;
|
||
}
|
||
|
||
private static void RequireAuditedVersion(string version)
|
||
{
|
||
if (!NativeSendPacket.IsAuditedVersion(version))
|
||
throw new InvalidOperationException($"主动发包编码器仅适配已核对版本 {NativeSendPacket.AuditedClientVersion}，当前为 {version}");
|
||
}
|
||
|
||
private static int RequireOpcode(OpcodeMap map, string version, PacketDirection direction, string name) =>
|
||
map.ResolveOpcode(version, direction, name)
|
||
?? throw new InvalidOperationException($"未找到 {version} 的 {name} Opcode 映射，禁止发送未映射消息");
|
||
}
|
||
|
||
public sealed class ActivePacketDispatcher(
|
||
string clientVersion,
|
||
Func<long>? clockMs = null,
|
||
int maxQueue = 32,
|
||
long minIntervalMs = 100,
|
||
long queueTimeoutMs = 2000,
|
||
long responseTimeoutMs = 5000)
|
||
{
|
||
private sealed class Transaction(long id, EncodedGamePacket packet, SendSession session, long createdMs)
|
||
{
|
||
public readonly long Id = id;
|
||
public readonly EncodedGamePacket Packet = packet;
|
||
public readonly SendSession Session = session;
|
||
public readonly long CreatedMs = createdMs;
|
||
public SendTransactionState State = SendTransactionState.Queued;
|
||
public long? SentMs;
|
||
public long? CompletedMs;
|
||
public byte? NativeResult;
|
||
public long? ResponseObservationId;
|
||
public long? RoundTripMs;
|
||
public string Detail = "排队等待主线程调度";
|
||
public SendTransactionSnapshot Snapshot() => new(
|
||
Id, Packet.Name, Packet.Opcode, State, CreatedMs, SentMs, CompletedMs,
|
||
NativeResult, ResponseObservationId, RoundTripMs, Detail);
|
||
}
|
||
|
||
[ThreadStatic] private static bool inReceiveCallback;
|
||
private readonly object sync = new();
|
||
private readonly Func<long> nowMs = clockMs ?? (() => Environment.TickCount64);
|
||
private readonly List<Transaction> history = [];
|
||
private SendSession lastSession = SendSession.Empty;
|
||
private long nextId;
|
||
private long lastSendMs = long.MinValue / 2;
|
||
|
||
public int PendingQueueCount { get { lock (sync) return history.Count(t => t.State == SendTransactionState.Queued); } }
|
||
public int InFlightCount { get { lock (sync) return history.Count(t => t.State == SendTransactionState.SentNative && !t.CompletedMs.HasValue); } }
|
||
|
||
public IDisposable Attach(NetworkPipeline pipeline) =>
|
||
pipeline.Subscribe(NetworkPhase.Received, OnReceived);
|
||
|
||
public SendTransactionSnapshot Enqueue(EncodedGamePacket packet, SendSession session)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(packet);
|
||
if (!NativeSendPacket.IsAuditedVersion(clientVersion))
|
||
throw new InvalidOperationException($"主动发送仅适配已核对版本 {NativeSendPacket.AuditedClientVersion}");
|
||
if (!session.IsValid)
|
||
throw new InvalidOperationException("当前角色或区域会话未就绪，禁止入队发送");
|
||
if (packet.Payload == null || packet.NativeBuffer == null ||
|
||
packet.NativeBuffer.Length != NativeSendPacket.NativePrefixSize + packet.Payload.Length)
|
||
throw new InvalidDataException("发送包原生缓冲区长度与载荷不匹配");
|
||
if (!PacketParser.Parse(packet.Name, packet.Payload, clientVersion).Parsed)
|
||
throw new InvalidDataException($"{packet.Name} 载荷未通过本机协议边界校验");
|
||
|
||
lock (sync)
|
||
{
|
||
var queuedCount = history.Count(t => t.State == SendTransactionState.Queued);
|
||
if (queuedCount >= maxQueue)
|
||
throw new InvalidOperationException($"发送队列已满（上限 {maxQueue} 条）");
|
||
if (lastSession.IsValid && !lastSession.Matches(session))
|
||
CancelSessionLocked("切区或角色会话变更，已取消旧会话请求");
|
||
lastSession = session;
|
||
var tx = new Transaction(++nextId, packet, session, nowMs());
|
||
history.Add(tx);
|
||
TrimHistoryLocked();
|
||
return tx.Snapshot();
|
||
}
|
||
}
|
||
|
||
// Canceling a queued item prevents native dispatch; it never claims to undo an already-sent packet.
|
||
public bool Cancel(long transactionId, string reason = "已取消排队请求")
|
||
{
|
||
lock (sync)
|
||
{
|
||
var tx = history.FirstOrDefault(t => t.Id == transactionId);
|
||
if (tx == null || tx.State != SendTransactionState.Queued) return false;
|
||
tx.State = SendTransactionState.Canceled;
|
||
tx.CompletedMs = nowMs();
|
||
tx.Detail = reason;
|
||
return true;
|
||
}
|
||
}
|
||
|
||
public int CancelAllQueued(string reason = "已清空待发送队列")
|
||
{
|
||
lock (sync)
|
||
{
|
||
var count = 0;
|
||
var now = nowMs();
|
||
foreach (var tx in history.Where(t => t.State == SendTransactionState.Queued))
|
||
{
|
||
tx.State = SendTransactionState.Canceled;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = reason;
|
||
count++;
|
||
}
|
||
return count;
|
||
}
|
||
}
|
||
|
||
// Must be called from Framework.Update on the main thread, never from a network receive callback.
|
||
public SendTransactionSnapshot? Pump(SendSession currentSession, Func<ReadOnlySpan<byte>, byte> nativeSender)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(nativeSender);
|
||
if (inReceiveCallback || NetworkPipeline.InCallback)
|
||
throw new InvalidOperationException("禁止在网络接收回调中直接触发主动发送");
|
||
|
||
Transaction? toSend = null;
|
||
var now = nowMs();
|
||
lock (sync)
|
||
{
|
||
if (!currentSession.IsValid || (lastSession.IsValid && !lastSession.Matches(currentSession)))
|
||
{
|
||
CancelSessionLocked("切区或会话不可用，已取消未完成请求");
|
||
lastSession = currentSession.IsValid ? currentSession : SendSession.Empty;
|
||
if (!currentSession.IsValid) return null;
|
||
}
|
||
else
|
||
{
|
||
lastSession = currentSession;
|
||
}
|
||
|
||
foreach (var tx in history)
|
||
{
|
||
if (tx.State == SendTransactionState.Queued && !tx.Session.Matches(currentSession))
|
||
{
|
||
tx.State = SendTransactionState.Canceled;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = "会话已变更，过期请求未发送";
|
||
}
|
||
else if (tx.State == SendTransactionState.Queued && now - tx.CreatedMs > queueTimeoutMs)
|
||
{
|
||
tx.State = SendTransactionState.TimedOut;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = $"排队超过 {queueTimeoutMs} ms 未发送，已过期丢弃";
|
||
}
|
||
else if (tx.State == SendTransactionState.SentNative && tx.Packet.ExpectedResponseOpcode.HasValue &&
|
||
tx.SentMs.HasValue && now - tx.SentMs.Value > responseTimeoutMs)
|
||
{
|
||
tx.State = SendTransactionState.TimedOut;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = $"已发送（原生返回 {tx.NativeResult}），但等待响应 0x{tx.Packet.ExpectedResponseOpcode.Value:X4} 超过 {responseTimeoutMs} ms";
|
||
}
|
||
}
|
||
|
||
if (now - lastSendMs < minIntervalMs) return null;
|
||
toSend = history.FirstOrDefault(t => t.State == SendTransactionState.Queued);
|
||
if (toSend == null) return null;
|
||
lastSendMs = now;
|
||
}
|
||
|
||
// Invoke native sender outside lock and outside observer catch blocks: never retry a native send.
|
||
byte nativeResult;
|
||
using (ActiveSendScope.Enter())
|
||
{
|
||
nativeResult = nativeSender(toSend.Packet.NativeBuffer);
|
||
}
|
||
|
||
var sentAt = nowMs();
|
||
lock (sync)
|
||
{
|
||
toSend.SentMs = sentAt;
|
||
toSend.NativeResult = nativeResult;
|
||
if (nativeResult == 0)
|
||
{
|
||
toSend.State = SendTransactionState.RejectedNative;
|
||
toSend.CompletedMs = sentAt;
|
||
toSend.Detail = "原生发送返回 0（连接未就绪或被底层拒绝）";
|
||
}
|
||
else if (toSend.Packet.ExpectedResponseOpcode.HasValue)
|
||
{
|
||
toSend.State = SendTransactionState.SentNative;
|
||
toSend.Detail = $"已调原生发送（返回 {nativeResult}），等待响应 0x{toSend.Packet.ExpectedResponseOpcode.Value:X4}";
|
||
}
|
||
else
|
||
{
|
||
toSend.State = SendTransactionState.SentNative;
|
||
toSend.CompletedMs = sentAt;
|
||
toSend.Detail = $"已调原生发送（返回 {nativeResult}，无回包契约，不代表服务端已执行）";
|
||
}
|
||
return toSend.Snapshot();
|
||
}
|
||
}
|
||
|
||
public SendTransactionSnapshot[] Snapshot()
|
||
{
|
||
lock (sync) return history.Select(t => t.Snapshot()).ToArray();
|
||
}
|
||
|
||
private void OnReceived(NetworkObservation observation)
|
||
{
|
||
var msg = observation.Message;
|
||
if (!msg.Complete || msg.Direction != PacketDirection.Down) return;
|
||
var prev = inReceiveCallback;
|
||
inReceiveCallback = true;
|
||
try
|
||
{
|
||
var now = nowMs();
|
||
lock (sync)
|
||
{
|
||
foreach (var tx in history)
|
||
{
|
||
if (tx.State != SendTransactionState.SentNative || !tx.Packet.ExpectedResponseOpcode.HasValue) continue;
|
||
if (tx.Packet.ExpectedResponseOpcode.Value != msg.Opcode) continue;
|
||
if (tx.SentMs.HasValue && now - tx.SentMs.Value > responseTimeoutMs)
|
||
{
|
||
tx.State = SendTransactionState.TimedOut;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = $"等待响应超时（>{responseTimeoutMs} ms）";
|
||
continue;
|
||
}
|
||
if (tx.Packet.ResponseMatcher != null && !tx.Packet.ResponseMatcher(msg.Payload)) continue;
|
||
tx.State = SendTransactionState.Confirmed;
|
||
tx.CompletedMs = now;
|
||
tx.ResponseObservationId = msg.Id;
|
||
tx.RoundTripMs = tx.SentMs.HasValue ? Math.Max(0, now - tx.SentMs.Value) : 0;
|
||
tx.Detail = $"收到响应确认（观察 #{msg.Id}，耗时 {tx.RoundTripMs} ms）";
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
finally
|
||
{
|
||
inReceiveCallback = prev;
|
||
}
|
||
}
|
||
|
||
private void CancelSessionLocked(string reason)
|
||
{
|
||
var now = nowMs();
|
||
foreach (var tx in history)
|
||
{
|
||
if (tx.State == SendTransactionState.Queued)
|
||
{
|
||
tx.State = SendTransactionState.Canceled;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = reason;
|
||
}
|
||
else if (tx.State == SendTransactionState.SentNative && !tx.CompletedMs.HasValue)
|
||
{
|
||
tx.State = SendTransactionState.Canceled;
|
||
tx.CompletedMs = now;
|
||
tx.Detail = "切区或会话变更，停止等待旧会话响应";
|
||
}
|
||
}
|
||
}
|
||
|
||
private void TrimHistoryLocked()
|
||
{
|
||
while (history.Count > 128)
|
||
{
|
||
var index = history.FindIndex(t => t.State != SendTransactionState.Queued);
|
||
if (index < 0) break;
|
||
history.RemoveAt(index);
|
||
}
|
||
}
|
||
}
|