diff --git a/Libraries/Esiur/Protocol/EpConnection.cs b/Libraries/Esiur/Protocol/EpConnection.cs index cf8ef05..12fa7ce 100644 --- a/Libraries/Esiur/Protocol/EpConnection.cs +++ b/Libraries/Esiur/Protocol/EpConnection.cs @@ -3549,29 +3549,22 @@ public partial class EpConnection : NetworkConnection, IStore foreach (var r in toBeRestored) { - var link = DC.ToBytes(r.ResourceLink); - Global.Log("EpConnection", LogType.Debug, "Restoreing " + r.ResourceLink); try { - var id = (uint)await SendRequest(EpPacketRequest.GetResourceIdByLink, link); - - - // remove from suspended. + // remove from suspended — Reattach(link, ...) below re-resolves + // and re-registers it under whatever id the server currently + // has (not necessarily the one we last knew — see Reattach's + // doc comment: the remote node's instance id is not permanent, + // only the link is), in a single round trip. _suspendedResources.Remove(r.ResourceInstanceId); - // id changed ? - if (id != r.ResourceInstanceId) - r.ResourceInstanceId = id; - - _neededResources[id] = r; - // Reattach using the last-known age so only properties modified while // disconnected are transferred and merged, instead of re-fetching all. - await Reattach(id, r.Instance.Age, r); + await Reattach(r.ResourceLink, r.Instance.Age, r); - Global.Log("EpConnection", LogType.Debug, "Restored " + id); + Global.Log("EpConnection", LogType.Debug, "Restored " + r.ResourceInstanceId); } catch (AsyncException ex) @@ -3720,6 +3713,14 @@ public partial class EpConnection : NetworkConnection, IStore _suspendedResources[r.ResourceInstanceId] = x; } } + // Every entry just moved to _suspendedResources — clear the old map too, + // otherwise Reattach()'s "already attached" fast path + // (`_attachedResources[id]?.TryGetTarget(...)`) finds the still-live + // weak reference to the now-suspended resource and returns it + // immediately, short-circuiting before _Reattach() ever runs — so a + // reconnect never actually resets Status back to Attached/Published + // or resubscribes events. + _attachedResources.Clear(); if (Server != null) { diff --git a/Libraries/Esiur/Protocol/EpConnectionProtocol.cs b/Libraries/Esiur/Protocol/EpConnectionProtocol.cs index 67b9980..50e25ac 100644 --- a/Libraries/Esiur/Protocol/EpConnectionProtocol.cs +++ b/Libraries/Esiur/Protocol/EpConnectionProtocol.cs @@ -1053,71 +1053,85 @@ partial class EpConnection void EpRequestReattachResource(uint callback, PlainTdu tdu) { - // resourceId, index, value + // resourceLinkOrId (string link, or uint id — a link is resolved by + // query since a remote node's instance id is not permanent: the + // resource may have been cleared from memory and recreated under a + // different id while the peer was disconnected), age var (valueOffset, valueSize, args) = DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, tdu.PayloadLength, Instance.Warehouse, 2); - var resourceId = Convert.ToUInt32(args[0]); - var age = Convert.ToUInt64(args[1]); - if (!TryBeginPeerAttachment(resourceId, out var exceptionCode, out var exceptionMessage)) + void Resolved(IResource res) { - SendError(ErrorType.Management, callback, (ushort)exceptionCode, exceptionMessage); - return; - } + if (res == null) + { + Global.Log("EpConnection", LogType.Debug, "Not found (reattach) " + args[0]); + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ResourceNotFound); + return; + } + + var resourceId = res.Instance.Id; + + if (!TryBeginPeerAttachment(resourceId, out var exceptionCode, out var exceptionMessage)) + { + SendError(ErrorType.Management, callback, (ushort)exceptionCode, exceptionMessage); + return; + } - Instance.Warehouse.GetById(resourceId).Then((res) => - { try { - if (res != null) - { - if (!TryApplyManagers( - null, - res, - ActionType.Attach, - callback, - ErrorType.Management, - ExceptionCode.AttachDenied, - out _)) - return; + if (!TryApplyManagers( + null, + res, + ActionType.Attach, + callback, + ErrorType.Management, + ExceptionCode.AttachDenied, + out _)) + return; - var r = res as IResource; + var r = res; - // unsubscribe - Unsubscribe(r); + // unsubscribe + Unsubscribe(r); + // reply ok — the resolved id comes first so the caller can + // detect and apply an id change (link-based reattach, or the + // remote node recreated the resource with a new id). + SendReply(EpPacketReply.Completed, callback, + r.Instance.Id, + r.Instance.Definition.Id, + r.Instance.Age, + r.Instance.Link, + r.Instance.Hops, + r.Instance.SerializeAfter(age)); - // reply ok - SendReply(EpPacketReply.Completed, callback, - r.Instance.Definition.Id, - r.Instance.Age, - r.Instance.Link, - r.Instance.Hops, - r.Instance.SerializeAfter(age)); - - - // subscribe - Subscribe(r); - } - else - { - // reply failed - Global.Log("EpConnection", LogType.Debug, "Not found " + resourceId); - SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ResourceNotFound); - } + // subscribe + Subscribe(r); } finally { EndPeerAttachment(resourceId); } - }).Error(ex => - { - EndPeerAttachment(resourceId); + } + + void Failed(Exception ex) => SendError(ErrorType.Management, callback, (ushort)ExceptionCode.GeneralFailure, ex.Message); - }); + + if (args[0] is string link) + { + if (Server?.EntryPoint != null) + Server.EntryPoint.Query(link, this).Then(Resolved).Error(Failed); + else + Instance.Warehouse.Query(link).Then(Resolved).Error(Failed); + } + else + { + var resourceId = Convert.ToUInt32(args[0]); + Instance.Warehouse.GetById(resourceId).Then(Resolved).Error(Failed); + } } void EpRequestDetachResource(uint callback, PlainTdu tdu) @@ -1496,7 +1510,11 @@ partial class EpConnection out _)) return; - SendReply(EpPacketReply.Completed, callback, r); + // Only the id — this opcode resolves a link to an identity, + // it does not attach. (A resource reference in a reply would + // auto-attach on decode, which is not what a bare id lookup + // should do — see Get(string) for the real attach step.) + SendReply(EpPacketReply.Completed, callback, r.Instance.Id); } }; @@ -2742,36 +2760,24 @@ partial class EpConnection /// Resource public AsyncReply Get(string path) { - var rt = new AsyncReply(); - var req = SendRequest(EpPacketRequest.GetResourceIdByLink, path); - - - req.Then(result => + // Two steps, deliberately: resolve the link to its (not permanent — + // the remote node may recreate the resource under a different id) + // current instance id, then attach by that id. GetResourceIdByLink + // returns a bare id, not a resource — it must not have the side + // effect of attaching anything on its own. + SendRequest(EpPacketRequest.GetResourceIdByLink, path).Then(result => + { + var id = Convert.ToUInt32(result); + FetchResource(id, null).Then(resource => { - // The resource is being handed to the application: remember it and publish its graph - // once all reachable dependencies have attached. - if (result is EpResource resource) - TrackDeliveredRoot(resource); - - rt.Trigger(result); + // The resource is being handed to the application: remember it and publish its + // graph once all reachable dependencies have attached. + TrackDeliveredRoot(resource); + rt.Trigger(resource); }).Error(ex => rt.TriggerError(ex)); - - - //Query(path).Then(ar => - //{ - - // //if (filter != null) - // // ar = ar?.Where(filter).ToArray(); - - // // MISSING: should dispatch the unused resources. - // if (ar?.Length > 0) - // rt.Trigger(ar[0]); - // else - // rt.Trigger(null); - //}).Error(ex => rt.TriggerError(ex)); - + }).Error(ex => rt.TriggerError(ex)); return rt; } @@ -3265,62 +3271,123 @@ partial class EpConnection /// merged into the existing instance instead of re-fetching everything. Falls back to a full /// if there is no prior state to merge into. /// - public AsyncReply Reattach(uint id, ulong age, EpResource resource) + /// + /// A link or a id. Prefer the link: the remote node's + /// instance id is not permanent (the resource may have been cleared from memory and recreated + /// under a different id while disconnected), but the link is — passing it resolves the current + /// id and reattaches in a single round trip instead of a separate GetResourceIdByLink first. + /// + public AsyncReply Reattach(object resourceLinkOrId, ulong age, EpResource resource) { - EpResource attachedResource = null; - _attachedResources[id]?.TryGetTarget(out attachedResource); - if (attachedResource != null) - return new AsyncReply(attachedResource); + // The already-attached / already-in-flight fast paths only apply when we + // already know a specific id — a link's current id isn't known until the + // server resolves it. (Concurrent duplicate reattach-by-link calls for the + // same resource aren't deduplicated the way FetchResource dedupes concurrent + // attach-by-id calls; the sole caller today — the reconnect loop — only ever + // calls this once per resource per reconnect, so this is a deliberate + // simplification, not an oversight.) + if (resourceLinkOrId is uint knownId) + { + EpResource attachedResource = null; + _attachedResources[knownId]?.TryGetTarget(out attachedResource); + if (attachedResource != null) + return new AsyncReply(attachedResource); - var existing = _resourceRequests[id]; - if (existing != null) - return existing.Reply; + var existing = _resourceRequests[knownId]; + if (existing != null) + return existing.Reply; + } var reply = new AsyncReply(); - var sequence = new uint[] { id }; - if (!TryReserveResourceAttachment(id, reply, sequence, out var alternative)) - return alternative; - ResourceAttachRequestCount++; - SendRequest(EpPacketRequest.ReattachResource, id, age).Then(result => + SendRequest(EpPacketRequest.ReattachResource, resourceLinkOrId, age).Then(result => { if (result == null) { - _resourceRequests.Remove(id); reply.TriggerError(new AsyncException(ErrorType.Management, (ushort)ExceptionCode.ResourceNotFound, "Null response")); return; } - // typeId, age, link, hops, delta(index -> PropertyValue) + // resolvedId, typeId, age, link, hops, delta(index -> PropertyValue) var args = (object[])result; - var deltaData = (byte[])args[4]; + var resolvedId = Convert.ToUInt32(args[0]); + var deltaData = (byte[])args[5]; + var sequence = new uint[] { resolvedId }; + var oldId = resource.ResourceInstanceId; DataDeserializer.PropertyValueMapParserAsync(deltaData, 0, (uint)deltaData.Length, this, sequence) .Then(delta => { + if (resolvedId != oldId) + { + // Re-key our own tracking to the id the server actually + // resolved the link to — see the id-is-not-permanent note above. + resource.ResourceInstanceId = resolvedId; + + // Only evict oldId's entries if they still point to THIS + // resource. If the remote node reused oldId for a different + // resource (freed, then reassigned) and something else + // concurrently attached it while this reattach was in + // flight, a blind Remove(oldId) would wrongly evict that + // unrelated resource's own valid tracking. + var oldAttachedRef = _attachedResources[oldId]; + if (oldAttachedRef != null + && oldAttachedRef.TryGetTarget(out var oldAttached) + && ReferenceEquals(oldAttached, resource)) + _attachedResources.Remove(oldId); + + if (ReferenceEquals(_neededResources[oldId], resource)) + _neededResources.Remove(oldId); + + // Safe unconditionally: this dictionary just marks "still + // suspended, needs restoring", which is false either way now + // that we're mid-reattach — no risk of evicting something + // unrelated's meaningful state. (Already removed by the + // reconnect loop's own pre-cleanup in the common case.) + _suspendedResources.Remove(oldId); + + // Reattach() never reserves a _resourceRequests entry before + // sending the request (see the fast-path comment above), so + // there is nothing of ours to remove here — left out + // entirely rather than risk canceling an unrelated in-flight + // request that happens to share the reused oldId. + } + if (!resource._Reattach(delta)) { // No prior state to merge into — perform a full attach instead. - _resourceRequests.Remove(id); - FetchResource(id, null).Then(r => reply.Trigger(r)).Error(ex => reply.TriggerError(ex)); + _resourceRequests.Remove(resolvedId); + FetchResource(resolvedId, null).Then(r => reply.Trigger(r)).Error(ex => reply.TriggerError(ex)); return; } - _resourceRequests.Remove(id); - _neededResources.Remove(id); - _attachedResources[id] = new WeakReference(resource); - ClearResourceFetchNode(id); + _resourceRequests.Remove(resolvedId); + _neededResources.Remove(resolvedId); + _attachedResources[resolvedId] = new WeakReference(resource); + ClearResourceFetchNode(resolvedId); + // TryPublishDeliveredRoots() alone isn't enough here: it only + // re-checks resources tracked in _deliveredRoots, which a + // resource that attached cleanly on its *first* attempt + // (the common case) never enters — see TrackDeliveredRoot's + // early return. Without this, a reattached resource stays + // stuck at ResourceStatus.Attached forever after a + // reconnect, since nothing else ever calls Publish() on it. + PublishGraph(resource); TryPublishDeliveredRoots(); reply.Trigger(resource); }) - .Error(ex => { _resourceRequests.Remove(id); ClearResourceFetchNode(id); reply.TriggerError(ex); }); - }).Error(ex => - { - _resourceRequests.Remove(id); - ClearResourceFetchNode(id); - reply.TriggerError(ex); - }); + .Error(ex => + { + _resourceRequests.Remove(resolvedId); + ClearResourceFetchNode(resolvedId); + reply.TriggerError(ex); + }); + }).Error(ex => reply.TriggerError(ex)); + // No _resourceRequests/fetch-node cleanup needed here: unlike + // FetchResource's id-based path, this method never reserves an + // entry before the request is sent (see the fast-path comment + // above), so there is nothing to unwind if the request itself fails. return reply; } diff --git a/Libraries/Esiur/Protocol/EpResource.cs b/Libraries/Esiur/Protocol/EpResource.cs index fb350c1..4e61d11 100644 --- a/Libraries/Esiur/Protocol/EpResource.cs +++ b/Libraries/Esiur/Protocol/EpResource.cs @@ -80,6 +80,18 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn EpResourceEvent[] _events; + // `On`/`Off` listener bags, keyed by property/event index — separate from + // the single-slot `_events[]` dynamic-member mechanism (`resource.Foo += + // handler`) above, which only ever holds one delegate per event. + readonly Dictionary>> _propertyListeners = new(); + readonly Dictionary>> _eventListeners = new(); + // Events we believe the server currently has us subscribed to — checked + // before sending a Subscribe/Unsubscribe request, since the server errors + // (AlreadyListened/AlreadyUnsubscribed) on a redundant one. + readonly HashSet _subscribedEvents = new(); + // Event indices with a subscription-reconciliation loop currently running. + readonly HashSet _reconciling = new(); + /// @@ -260,6 +272,12 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn } _status = Resource.ResourceStatus.Attached; + // A reattach can follow an unexpected disconnect + automatic + // reconnect: the server-side subscription state keyed to the old, + // now-dead connection is gone, but our local listeners (On()/+=) + // are untouched, so without this we'd wrongly believe we're still + // subscribed and never resend the wire Subscribe request. + ReconcileAllSubscriptions(); return true; } @@ -269,6 +287,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn var et = Instance.Definition.GetEventDefByIndex(index); _events[index]?.Invoke(this, args); Instance.EmitResourceEvent(et, args); + DispatchListeners(_eventListeners, index, args); } public AsyncReply _Invoke(byte index, object args) @@ -371,6 +390,164 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn return Unsubscribe(et); } + /// + /// Listen for a property change (On(":propName", cb)) or an + /// exported event (On("eventName", cb)). For events where the + /// TypeDef marks (i.e. not + /// [AutoDelivery]), the first listener triggers a Subscribe + /// request and the last triggers Unsubscribe — + /// ref-counted by listener count, so redundant wire requests aren't sent + /// for a second/third listener on the same already-subscribed event. + /// + public EpResource On(string name, Action callback) + { + if (name.StartsWith(":")) + { + var propertyName = name.Substring(1); + var pt = Instance.Definition.GetPropertyDefByName(propertyName) + ?? throw new Exception($"Unknown property \"{propertyName}\"."); + AddListener(_propertyListeners, pt.Index, callback); + return this; + } + + var et = Instance.Definition.GetEventDefByName(name) + ?? throw new Exception($"Unknown event \"{name}\"."); + AddListener(_eventListeners, et.Index, callback); + if (et.Subscribable) ReconcileSubscription(et); + return this; + } + + /// Remove a listener registered with . + public EpResource Off(string name, Action callback) + { + if (name.StartsWith(":")) + { + var pt = Instance.Definition.GetPropertyDefByName(name.Substring(1)); + if (pt != null) RemoveListener(_propertyListeners, pt.Index, callback); + return this; + } + + var et = Instance.Definition.GetEventDefByName(name); + if (et == null) return this; + RemoveListener(_eventListeners, et.Index, callback); + if (et.Subscribable) ReconcileSubscription(et); + return this; + } + + static void AddListener(Dictionary>> map, byte index, Action callback) + { + lock (map) + { + if (!map.TryGetValue(index, out var list)) + { + list = new List>(); + map[index] = list; + } + list.Add(callback); + } + } + + static void RemoveListener(Dictionary>> map, byte index, Action callback) + { + lock (map) + if (map.TryGetValue(index, out var list)) + list.Remove(callback); + } + + static int ListenerCount(Dictionary>> map, byte index) + { + lock (map) + return map.TryGetValue(index, out var list) ? list.Count : 0; + } + + static void DispatchListeners(Dictionary>> map, byte index, object value) + { + Action[] callbacks; + lock (map) + { + if (!map.TryGetValue(index, out var list) || list.Count == 0) return; + callbacks = list.ToArray(); + } + foreach (var callback in callbacks) + callback(value); + } + + /// + /// Called from : reset our belief about server-side + /// subscription state (a fresh connection starts with none) and re-run + /// reconciliation for every subscribable event that still has active + /// listeners, so subscriptions survive an unexpected disconnect + + /// automatic reconnect transparently. + /// + void ReconcileAllSubscriptions() + { + lock (_subscribedEvents) _subscribedEvents.Clear(); + lock (_reconciling) _reconciling.Clear(); + + foreach (var et in Instance.Definition.Events) + { + if (!et.Subscribable) continue; + var hasListeners = ListenerCount(_eventListeners, et.Index) > 0 || _events[et.Index] != null; + if (hasListeners) ReconcileSubscription(et); + } + } + + /// + /// Settle the wire subscription state for toward + /// whatever the current listener count implies, rechecking the *current* + /// desired state on each step — so a burst of On()/Off() calls while a + /// request is in flight is coalesced into whatever the state actually is + /// once the in-flight request settles, rather than replaying every + /// transition. + /// + void ReconcileSubscription(EventDef et) + { + lock (_reconciling) + { + if (!_reconciling.Add(et.Index)) return; + } + StepSubscription(et); + } + + void StepSubscription(EventDef et) + { + bool desired; + bool actual; + lock (_subscribedEvents) + { + // Either subscription mechanism wanting delivery is enough: + // `On()`/`Off()`'s ref-counted list, or a native `+=`/`-=` + // multicast delegate assigned via TrySetMember. + desired = ListenerCount(_eventListeners, et.Index) > 0 || _events[et.Index] != null; + actual = _subscribedEvents.Contains(et.Index); + } + + if (desired == actual) + { + lock (_reconciling) _reconciling.Remove(et.Index); + return; + } + + var request = desired ? Subscribe(et) : Unsubscribe(et); + request.Then((_) => + { + lock (_subscribedEvents) + { + if (desired) _subscribedEvents.Add(et.Index); + else _subscribedEvents.Remove(et.Index); + } + // Re-check: the desired state may have changed while this request + // was in flight (another On()/Off() call came in meanwhile). + StepSubscription(et); + }).Error((_) => + { + // Leave `_subscribedEvents` as-is; the next On()/Off() call that + // changes the listener count re-triggers reconciliation, so a + // transient failure here just needs another transition to retry. + lock (_reconciling) _reconciling.Remove(et.Index); + }); + } + public override bool TryInvokeMember(InvokeMemberBinder binder, object[] args, out object result) { @@ -492,6 +669,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn var pt = Instance.Definition.GetPropertyDefByIndex(index); _properties[index] = value; Instance.EmitModification(pt, value); + DispatchListeners(_propertyListeners, index, value); } /// @@ -538,7 +716,17 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn if (et == null) return false; + // `r.Milestone += handler` desugars to a get (TryGetMember) + a + // DLR-computed `Delegate.Combine` + this set — so this slot is + // already a proper multicast delegate across multiple `+=` + // subscribers. What's new is noticing the null <-> non-null + // transition to (un)subscribe on the wire, matching `On`/`Off`. + var wasEmpty = _events[et.Index] == null; _events[et.Index] = (EpResourceEvent)value; + var isEmpty = _events[et.Index] == null; + + if (wasEmpty != isEmpty && et.Subscribable) + ReconcileSubscription(et); return true; } diff --git a/Tests/Unit/Integration/BeaconResource.cs b/Tests/Unit/Integration/BeaconResource.cs new file mode 100644 index 0000000..5679e04 --- /dev/null +++ b/Tests/Unit/Integration/BeaconResource.cs @@ -0,0 +1,29 @@ +using Esiur.Resource; + +namespace Esiur.Tests.Unit.Integration; + +[Resource] +public partial class BeaconResource +{ + [Export] int pings; + + // No [AutoDelivery]: subscribable by default — requires an explicit + // Subscribe request before occurrences flow to a given connection. + [Export] public event ResourceEventHandler? Ping; + + // Opts out via [AutoDelivery]: flows to every attached connection + // unconditionally. + [Export] + [AutoDelivery] + public event ResourceEventHandler? Tick; + + public void Fire(string name, string value) + { + // Write through the generated public property (Pings), not the + // private backing field — only the generated setter emits the + // PropertyModified notification. + Pings = pings + 1; + if (name == "ping") Ping?.Invoke(value); + else if (name == "tick") Tick?.Invoke(value); + } +} diff --git a/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs b/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs new file mode 100644 index 0000000..44f76b9 --- /dev/null +++ b/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs @@ -0,0 +1,207 @@ +using Esiur.Protocol; +using Esiur.Resource; + +namespace Esiur.Tests.Unit.Integration; + +[Collection("Integration")] +public class EventSubscriptionIntegrationTests +{ + [Fact] + public async Task On_DeliversAutoDeliveredEventWithNoSubscribeNeeded() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + + var received = new List(); + remote.On("Tick", v => received.Add((string)v)); + + beacon.Fire("tick", "a"); + await WaitUntilAsync(() => received.Count == 1, TimeSpan.FromSeconds(3)); + + Assert.Equal(new[] { "a" }, received); + } + + [Fact] + public async Task On_RefCountsListeners_OffOnlyUnsubscribesAtZero() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + + var a = new List(); + var b = new List(); + void CbA(object v) => a.Add((string)v); + void CbB(object v) => b.Add((string)v); + + remote.On("Ping", CbA); + // Give the wire Subscribe request time to round-trip before firing. + await Task.Delay(200); + + remote.On("Ping", CbB); // 2nd listener on an already-subscribed event + await Task.Delay(100); + + beacon.Fire("ping", "x"); + await WaitUntilAsync(() => a.Count == 1 && b.Count == 1, TimeSpan.FromSeconds(3)); + + remote.Off("Ping", CbA); // one listener remains — must not unsubscribe yet + await Task.Delay(100); + + beacon.Fire("ping", "y"); + await WaitUntilAsync(() => b.Count == 2, TimeSpan.FromSeconds(3)); + Assert.Equal(new[] { "x" }, a); // CbA got nothing after being removed + + remote.Off("Ping", CbB); // last listener — now it should unsubscribe + await Task.Delay(200); + + beacon.Fire("ping", "z"); + await Task.Delay(200); + Assert.Equal(new[] { "x" }, a); + Assert.Equal(new[] { "x", "y" }, b); // neither received "z" + } + + [Fact] + public async Task NativePlusEquals_AutoSubscribesAndUnsubscribes() + { + // `r.Ping += handler` is the built-in, idiomatic C# way to listen — + // TrySetMember already combines this into a proper multicast + // delegate; what's new is that it now also drives Subscribe/Unsubscribe. + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + dynamic dyn = remote; + + var received = new List(); + EpResourceEvent handler = (_, arg) => received.Add((string)arg); + + dyn.Ping += handler; + await Task.Delay(200); // let the wire Subscribe request round-trip + + beacon.Fire("ping", "x"); + await WaitUntilAsync(() => received.Count == 1, TimeSpan.FromSeconds(3)); + Assert.Equal(new[] { "x" }, received); + + dyn.Ping -= handler; + await Task.Delay(200); // let the wire Unsubscribe request round-trip + + beacon.Fire("ping", "y"); + await Task.Delay(200); + Assert.Equal(new[] { "x" }, received); // "y" never arrives once unsubscribed + } + + [Fact] + public async Task On_AndNativePlusEquals_ComposeCorrectly() + { + // `On()`'s ref-counted list and `+=`'s multicast delegate are two + // independent subscriber sources for the same wire subscription — + // either being non-empty must keep it subscribed. + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + dynamic dyn = remote; + + var onReceived = new List(); + var plusReceived = new List(); + void OnCb(object v) => onReceived.Add((string)v); + EpResourceEvent plusHandler = (_, arg) => plusReceived.Add((string)arg); + + remote.On("Ping", OnCb); + await Task.Delay(200); + + dyn.Ping += plusHandler; // 2nd subscriber via the other mechanism + await Task.Delay(100); + + beacon.Fire("ping", "x"); + await WaitUntilAsync(() => onReceived.Count == 1 && plusReceived.Count == 1, TimeSpan.FromSeconds(3)); + + remote.Off("Ping", OnCb); // On()'s listener gone, but += subscriber remains + await Task.Delay(100); + + beacon.Fire("ping", "y"); + await WaitUntilAsync(() => plusReceived.Count == 2, TimeSpan.FromSeconds(3)); + Assert.Equal(new[] { "x" }, onReceived); // no "y" — Off() already removed it + + dyn.Ping -= plusHandler; // last subscriber — now it should unsubscribe + await Task.Delay(200); + + beacon.Fire("ping", "z"); + await Task.Delay(200); + Assert.Equal(new[] { "x", "y" }, plusReceived); // neither got "z" + } + + [Fact] + public async Task On_ResubscribesAfterAutomaticReconnect() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + cluster.Connection.AutoReconnect = true; + cluster.Connection.ReconnectInterval = 1; + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + + var received = new List(); + remote.On("Ping", v => received.Add((string)v)); + await Task.Delay(200); // let the initial Subscribe land + + beacon.Fire("ping", "before"); + await WaitUntilAsync(() => received.Count == 1, TimeSpan.FromSeconds(3)); + + // Simulate an unexpected disconnect — the server-side subscription + // state (keyed to the now-dead connection) is gone, but the client's + // local `On()` listener is untouched, so it still believes it's + // subscribed unless _Reattach's ReconcileAllSubscriptions() resets that. + foreach (var serverConnection in cluster.Server.Connections.ToArray()) + serverConnection.Destroy(); + + await WaitUntilAsync(() => !cluster.Connection.IsConnected, TimeSpan.FromSeconds(3)); + await WaitUntilAsync(() => cluster.Connection.IsConnected, TimeSpan.FromSeconds(5)); + await Task.Delay(300); // let the post-reattach resubscribe land + + beacon.Fire("ping", "after"); + await WaitUntilAsync(() => received.Count == 2, TimeSpan.FromSeconds(3)); + Assert.Equal(new[] { "before", "after" }, received); + } + + [Fact] + public async Task On_PropertyPrefix_ListensWithNoWireSubscription() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + + var seen = new List(); + remote.On(":Pings", v => seen.Add(v)); + + beacon.Fire("tick", "z"); + await WaitUntilAsync(() => seen.Count == 1, TimeSpan.FromSeconds(3)); + + Assert.Equal(1, Convert.ToInt32(seen[0])); + } + + static Task StartClusterAsync(out Func getBeacon) + { + BeaconResource? beacon = null; + var clusterTask = IntegrationCluster.StartAsync(async warehouse => + { + beacon = new BeaconResource(); + await warehouse.Put("sys/beacon", beacon); + }); + getBeacon = () => beacon!; + return clusterTask; + } + + static async Task GetRemote(IntegrationCluster cluster) + => (EpResource)await Task.Run(async () => + await cluster.Connection.Get("sys/beacon")) + .WaitAsync(TimeSpan.FromSeconds(10)); + + static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition()) + { + if (DateTime.UtcNow >= deadline) + throw new TimeoutException("The expected condition was not reached."); + await Task.Delay(20); + } + } +} diff --git a/Tools/Esiur.CLI/Generation/TypeScriptStubGenerator.cs b/Tools/Esiur.CLI/Generation/TypeScriptStubGenerator.cs index ee4947b..5cd6617 100644 --- a/Tools/Esiur.CLI/Generation/TypeScriptStubGenerator.cs +++ b/Tools/Esiur.CLI/Generation/TypeScriptStubGenerator.cs @@ -136,6 +136,16 @@ public static class TypeScriptStubGenerator sb.AppendLine($" {function.Name}({args}): PromiseLike<{returnType}>;"); } + // Events aren't interface members in TypeScript (no static member shape + // for `.on()`/`.off()` to type-check against yet) — documented as a + // comment instead so the event name/argument type are still discoverable. + foreach (var evt in typeDef.Events.OrderBy(e => e.Index)) + { + var argType = TypeScriptTypeMapper.Map(evt.ArgumentType, names); + var wire = evt.Subscribable ? "subscribable" : "auto-delivered"; + sb.AppendLine($" // event {evt.Name}: on(\"{evt.Name}\", (value: {argType}) => void) — {wire}"); + } + sb.AppendLine("}"); sb.AppendLine(); }