using System.Diagnostics; namespace ICRebuild.Core; public enum NetworkPhase { Received, BeforeSend, AfterSend } public enum PacketOrigin { NativeZone, PluginActiveSend } public enum PacketStage { RawIpcPayload } public sealed record PacketContext(long ObservationId, PacketOrigin Origin, PacketStage Stage, NetworkPhase Phase, byte? NativeResult = null); // Owned immutable copy. No native pointer or mutable array is exposed to observers. public sealed class NetworkMessage { private readonly byte[] payload; public long Id { get; } public DateTimeOffset Time { get; } = DateTimeOffset.UtcNow; public long MonotonicTicks { get; } = Stopwatch.GetTimestamp(); public PacketDirection Direction { get; } public string GameVersion { get; } public int Opcode { get; } public uint? Target { get; } public bool Complete { get; } public string? Issue { get; } public PacketOrigin Origin { get; } public ReadOnlySpan Payload => payload; public NetworkMessage(long id, PacketDirection direction, string version, CapturedPacket packet, PacketOrigin origin = PacketOrigin.NativeZone) { Id=id; Direction=direction; GameVersion=version; Opcode=packet.Opcode; Target=packet.Target; Complete=packet.Complete; Issue=packet.Issue; Origin=origin; payload=packet.Payload.ToArray(); } } public sealed record NetworkObservation(NetworkMessage Message, NetworkPhase Phase, byte? NativeResult = null) { public PacketContext Context => new(Message.Id,Message.Origin,PacketStage.RawIpcPayload,Phase,NativeResult); } // Inline on the native calling thread, in registration order per observation. Concurrent // calls may interleave. Callbacks must be short; no game/UI service access, waits for // Framework, native sending, or changes to native Hook lifetimes from a callback. public sealed class NetworkPipeline(string version) : IDisposable { private sealed class Subscription(NetworkPhase phase, Action callback, int? opcode) { public readonly NetworkPhase Phase=phase; public readonly Action Callback=callback; public readonly int? Opcode=opcode; public int Active=1; public bool Matches(NetworkPhase phase, int? opcode=null) => Volatile.Read(ref Active)!=0 && Phase==phase && (!opcode.HasValue || !Opcode.HasValue || Opcode==opcode); } private readonly object sync=new(); private Subscription[] subscriptions=[]; private bool disposed; private long sequence,readErrors,observerErrors,received,beforeSend,afterSend; private int lastSendResult=-1; private string lastIssue="无"; public long ReadErrors => Interlocked.Read(ref readErrors); public long ObserverErrors => Interlocked.Read(ref observerErrors); public long Received => Interlocked.Read(ref received); public long BeforeSend => Interlocked.Read(ref beforeSend); public long AfterSend => Interlocked.Read(ref afterSend); public int LastSendResult => Volatile.Read(ref lastSendResult); public string LastIssue => Volatile.Read(ref lastIssue); public int SubscriberCount => Volatile.Read(ref subscriptions).Length; public IDisposable Subscribe(NetworkPhase phase, Action callback, int? opcode=null) { ArgumentNullException.ThrowIfNull(callback); if(!Enum.IsDefined(phase) || opcode is <0 or >65535) throw new ArgumentOutOfRangeException(nameof(opcode)); var item=new Subscription(phase,callback,opcode); lock(sync) { ObjectDisposedException.ThrowIf(disposed,this); Volatile.Write(ref subscriptions,[..subscriptions,item]); } return new ActionDisposable(()=> { // Already-entered callbacks may finish; cancellation cannot recall them. Interlocked.Exchange(ref item.Active,0); lock(sync) Volatile.Write(ref subscriptions,subscriptions.Where(x=>x!=item).ToArray()); }); } public byte Send(Func original, Func,CaptureRead> capture) { var listeners=Volatile.Read(ref subscriptions); NetworkMessage? message=null; if(listeners.Any(s=>s.Matches(NetworkPhase.BeforeSend) || s.Matches(NetworkPhase.AfterSend))) message=Read(PacketDirection.Up,capture,opcode=>listeners.Any(s=> s.Matches(NetworkPhase.BeforeSend,opcode) || s.Matches(NetworkPhase.AfterSend,opcode))); if(message!=null) { Interlocked.Increment(ref beforeSend); Publish(listeners,new(message,NetworkPhase.BeforeSend)); } // Outside observer/read exception handlers: never retry the native call. var result=original(); Volatile.Write(ref lastSendResult,result); if(message!=null) { Interlocked.Increment(ref afterSend); Publish(listeners,new(message,NetworkPhase.AfterSend,result)); } return result; } public byte Receive(Func original, Func,CaptureRead> capture) { var result=original(); if(result==0) return result; // Holder is unreadable on unsuccessful receive. var listeners=Volatile.Read(ref subscriptions); if(listeners.Any(s=>s.Matches(NetworkPhase.Received))) { var message=Read(PacketDirection.Down,capture,opcode=>listeners.Any(s=>s.Matches(NetworkPhase.Received,opcode))); if(message!=null) { Interlocked.Increment(ref received); Publish(listeners,new(message,NetworkPhase.Received,result)); } } return result; } private NetworkMessage? Read(PacketDirection direction,Func,CaptureRead> capture,Func accepts) { try { var read=capture(accepts); if(read.Error!=null) { ReadFailed(read.Error); return null; } if(read.Packet is not {} packet) return null; if(!packet.Complete) ReadFailed(packet.Issue ?? "载荷读取不完整"); var origin=direction==PacketDirection.Up && ActiveSendScope.Active ? PacketOrigin.PluginActiveSend : PacketOrigin.NativeZone; return new(Interlocked.Increment(ref sequence),direction,version,packet,origin); } catch(Exception ex) { ReadFailed(ex.Message); return null; } } private void ReadFailed(string issue) { Interlocked.Increment(ref readErrors); Volatile.Write(ref lastIssue,issue); } [ThreadStatic] private static int callbackDepth; public static bool InCallback => callbackDepth > 0; private void Publish(Subscription[] listeners,NetworkObservation observation) { callbackDepth++; try { foreach(var listener in listeners) { if(!listener.Matches(observation.Phase,observation.Message.Opcode)) continue; try { listener.Callback(observation); } catch(Exception ex) { Interlocked.Increment(ref observerErrors); Volatile.Write(ref lastIssue,$"{observation.Phase} 订阅异常:{ex.Message}"); } } } finally { if(callbackDepth>0) callbackDepth--; } } public void Dispose() { lock(sync) { disposed=true; foreach(var listener in subscriptions) Interlocked.Exchange(ref listener.Active,0); Volatile.Write(ref subscriptions,[]); } } }