diff --git a/Libraries/Esiur/Core/AsyncReply.cs b/Libraries/Esiur/Core/AsyncReply.cs index 30dc518..872c2e1 100644 --- a/Libraries/Esiur/Core/AsyncReply.cs +++ b/Libraries/Esiur/Core/AsyncReply.cs @@ -271,7 +271,7 @@ public class AsyncReply return; } - public void TriggerError(Exception exception) + public virtual void TriggerError(Exception exception) { //timeout?.Dispose(); diff --git a/Libraries/Esiur/Core/AsyncStreamReply.cs b/Libraries/Esiur/Core/AsyncStreamReply.cs new file mode 100644 index 0000000..7dfb47b --- /dev/null +++ b/Libraries/Esiur/Core/AsyncStreamReply.cs @@ -0,0 +1,409 @@ +/* + +Copyright (c) 2017-2026 Ahmed Kh. Zamil + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + +*/ + +using Esiur.Data; +using Esiur.Data.Types; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Esiur.Core; + +/// +/// Represents a remotely executing stream and exposes its lifecycle controls. +/// +public class AsyncStreamReply : AsyncReply +{ + readonly Func _pull; + readonly Func _terminate; + readonly Func _halt; + readonly Func _resume; + + readonly object _streamLock = new object(); + readonly TaskCompletionSource _started = NewCompletionSource(); + + Exception _streamException; + bool _streamStarted; + bool _streamCompleted; + bool _terminationSent; + + /// + /// Gets the delivery mode declared by the remote function. + /// + public StreamMode StreamMode { get; } + + /// + /// Gets whether the peer acknowledged that the stream has started. + /// + public bool Started + { + get + { + lock (_streamLock) + return _streamStarted; + } + } + + /// + /// Gets whether the stream completed or was terminated. + /// + public bool Completed + { + get + { + lock (_streamLock) + return _streamCompleted; + } + } + + internal AsyncStreamReply( + StreamMode streamMode, + Func pull, + Func terminate, + Func halt, + Func resume) + { + StreamMode = streamMode; + _pull = pull; + _terminate = terminate; + _halt = halt; + _resume = resume; + } + + /// + /// Requests the next item of a pull stream. + /// + public AsyncReply Pull() + { + if (StreamMode != StreamMode.Pull) + throw new InvalidOperationException("Pull is only valid for a pull stream."); + + lock (_streamLock) + { + if (_streamCompleted) + return new AsyncReply(null); + } + + return _pull(); + } + + /// + /// Terminates the remote stream execution and releases its enumerator. + /// + public AsyncReply Terminate() + { + lock (_streamLock) + { + if (_terminationSent || _streamCompleted) + return new AsyncReply(null); + + _terminationSent = true; + _streamCompleted = true; + _started.TrySetResult(true); + } + + OnStreamCompleted(); + + var reply = _terminate(); + reply.Error(TriggerStreamError); + return reply; + } + + /// + /// Halts a pausable remote stream execution. + /// + public AsyncReply Halt() + { + lock (_streamLock) + { + if (_streamCompleted) + return new AsyncReply(null); + } + + var reply = _halt(); + reply.Error(TriggerStreamError); + return reply; + } + + /// + /// Resumes a halted remote stream execution. + /// + public AsyncReply Resume() + { + lock (_streamLock) + { + if (_streamCompleted) + return new AsyncReply(null); + } + + var reply = _resume(); + reply.Error(TriggerStreamError); + return reply; + } + + internal void TriggerStreamStarted() + { + lock (_streamLock) + { + if (_streamCompleted || _streamException != null) + return; + + _streamStarted = true; + _started.TrySetResult(true); + } + + OnStreamStarted(); + } + + internal void TriggerStreamCompleted() + { + lock (_streamLock) + { + if (_streamCompleted) + return; + + _streamCompleted = true; + _started.TrySetResult(true); + } + + OnStreamCompleted(); + } + + internal void TriggerStreamError(AsyncException exception) + { + lock (_streamLock) + { + if (_streamCompleted || _streamException != null) + return; + + _streamException = exception; + _streamCompleted = true; + _started.TrySetException(exception); + } + + OnStreamError(exception); + + if (errorCallbacks != null) + base.TriggerError(exception); + } + + /// + public override void TriggerError(Exception exception) + { + TriggerStreamError(exception as AsyncException ?? new AsyncException(exception)); + } + + internal Task WaitUntilStartedAsync() => _started.Task; + + internal Exception GetStreamException() + { + lock (_streamLock) + return _streamException; + } + + internal bool IsStreamCompleted() + { + lock (_streamLock) + return _streamCompleted; + } + + protected virtual void OnStreamStarted() { } + protected virtual void OnStreamCompleted() { } + protected virtual void OnStreamError(Exception exception) { } + + protected static TaskCompletionSource NewCompletionSource() + => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); +} + +/// +/// A typed remote stream that can be consumed with await foreach. +/// Pull streams request exactly one remote item for each MoveNextAsync call. +/// +public sealed class AsyncStreamReply : AsyncStreamReply, IAsyncEnumerable +{ + readonly object _itemsLock = new object(); + readonly Queue _items = new Queue(); + + TaskCompletionSource _itemAvailable; + bool _enumeratorCreated; + bool _movePending; + + internal AsyncStreamReply( + StreamMode streamMode, + Func pull, + Func terminate, + Func halt, + Func resume) + : base(streamMode, pull, terminate, halt, resume) + { + Chunk(value => ReceiveItem((T)RuntimeCaster.Cast(value, typeof(T)))); + } + + /// + public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = default) + { + lock (_itemsLock) + { + if (_enumeratorCreated) + throw new InvalidOperationException("A remote stream can only be enumerated once."); + + _enumeratorCreated = true; + } + + return new Enumerator(this, cancellationToken); + } + + void ReceiveItem(T item) + { + TaskCompletionSource itemAvailable; + + lock (_itemsLock) + { + if (IsStreamCompleted() || GetStreamException() != null) + return; + + _items.Enqueue(item); + itemAvailable = _itemAvailable; + _itemAvailable = null; + } + + itemAvailable?.TrySetResult(true); + } + + protected override void OnStreamCompleted() + { + TaskCompletionSource itemAvailable; + + lock (_itemsLock) + { + itemAvailable = _itemAvailable; + _itemAvailable = null; + } + + itemAvailable?.TrySetResult(true); + } + + protected override void OnStreamError(Exception exception) + { + TaskCompletionSource itemAvailable; + + lock (_itemsLock) + { + itemAvailable = _itemAvailable; + _itemAvailable = null; + } + + itemAvailable?.TrySetException(exception); + } + + async Task<(bool HasValue, T Value)> MoveNextAsync(CancellationToken cancellationToken) + { + lock (_itemsLock) + { + if (_movePending) + throw new InvalidOperationException("Concurrent MoveNextAsync calls are not supported."); + + _movePending = true; + } + + try + { + cancellationToken.ThrowIfCancellationRequested(); + await WaitUntilStartedAsync(); + + while (true) + { + TaskCompletionSource itemAvailable; + + lock (_itemsLock) + { + var exception = GetStreamException(); + if (exception != null) + throw exception; + + if (_items.Count > 0) + return (true, _items.Dequeue()); + + if (IsStreamCompleted()) + return (false, default); + + _itemAvailable = NewCompletionSource(); + itemAvailable = _itemAvailable; + } + + if (StreamMode == StreamMode.Pull) + { + var pull = Pull(); + _ = pull.Error(TriggerStreamError); + } + + using (cancellationToken.Register(() => + { + itemAvailable.TrySetCanceled(); + _ = Terminate(); + })) + { + await itemAvailable.Task; + } + } + } + finally + { + lock (_itemsLock) + _movePending = false; + } + } + + async Task DisposeAsync() + { + if (IsStreamCompleted()) + return; + + await Terminate(); + } + + sealed class Enumerator : IAsyncEnumerator + { + readonly AsyncStreamReply _owner; + readonly CancellationToken _cancellationToken; + + public T Current { get; private set; } + + internal Enumerator(AsyncStreamReply owner, CancellationToken cancellationToken) + { + _owner = owner; + _cancellationToken = cancellationToken; + } + + public async ValueTask MoveNextAsync() + { + var result = await _owner.MoveNextAsync(_cancellationToken); + Current = result.Value; + return result.HasValue; + } + + public ValueTask DisposeAsync() => new ValueTask(_owner.DisposeAsync()); + } +} diff --git a/Libraries/Esiur/Core/ExceptionCode.cs b/Libraries/Esiur/Core/ExceptionCode.cs index b5624fa..90323d4 100644 --- a/Libraries/Esiur/Core/ExceptionCode.cs +++ b/Libraries/Esiur/Core/ExceptionCode.cs @@ -46,5 +46,6 @@ public enum ExceptionCode : ushort Timeout, NotSupported, NotImplemented, - NotAllowed + NotAllowed, + RateLimitExceeded } diff --git a/Libraries/Esiur/Core/InvocationContext.cs b/Libraries/Esiur/Core/InvocationContext.cs index 9c5e847..73dbe73 100644 --- a/Libraries/Esiur/Core/InvocationContext.cs +++ b/Libraries/Esiur/Core/InvocationContext.cs @@ -1,15 +1,47 @@ -using Esiur.Protocol; +using Esiur.Data.Types; +using Esiur.Protocol; using System; +using System.Collections; using System.Collections.Generic; -using System.Text; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; namespace Esiur.Core { public class InvocationContext { - private uint CallbackId; + readonly uint CallbackId; + readonly object _stateLock = new object(); + readonly CancellationTokenSource _cancellation = new CancellationTokenSource(); + readonly SemaphoreSlim _moveLock = new SemaphoreSlim(1, 1); + TaskCompletionSource _resumeSignal = CompletedSignal(); - internal bool Ended; + IAsyncEnumeratorAdapter _asyncEnumerator; + IEnumerator _enumerator; + bool _halted; + + internal StreamMode StreamMode { get; private set; } + internal bool Pausable { get; private set; } + + internal volatile bool Ended; + + /// + /// Gets a token that is cancelled when the caller terminates this execution. + /// + public CancellationToken CancellationToken => _cancellation.Token; + + /// + /// Gets whether a pausable execution is currently halted. + /// + public bool IsHalted + { + get + { + lock (_stateLock) + return _halted; + } + } public void Chunk(object value) { @@ -19,8 +51,8 @@ namespace Esiur.Core Connection.SendChunk(CallbackId, value); } - public void Progress(uint value, uint max) { - + public void Progress(uint value, uint max) + { if (Ended) throw new Exception("Execution has ended."); @@ -29,7 +61,6 @@ namespace Esiur.Core public void Warning(byte level, string message) { - if (Ended) throw new Exception("Execution has ended."); @@ -38,12 +69,192 @@ namespace Esiur.Core public EpConnection Connection { get; internal set; } - internal InvocationContext(EpConnection connection, uint callbackId) { Connection = connection; CallbackId = callbackId; + } + internal void InitializeStream(StreamMode streamMode, bool pausable) + { + StreamMode = streamMode; + Pausable = pausable; + } + + internal bool SetAsyncEnumerable(object enumerable) + { + var asyncEnumerableInterface = enumerable.GetType() + .GetInterfaces() + .Concat(new[] { enumerable.GetType() }) + .FirstOrDefault(x => x.IsGenericType && + x.GetGenericTypeDefinition() == typeof(IAsyncEnumerable<>)); + + if (asyncEnumerableInterface == null) + return false; + + var itemType = asyncEnumerableInterface.GetGenericArguments()[0]; + var adapterType = typeof(AsyncEnumeratorAdapter<>).MakeGenericType(itemType); + _asyncEnumerator = (IAsyncEnumeratorAdapter)Activator.CreateInstance( + adapterType, enumerable, _cancellation.Token); + return true; + } + + internal void SetEnumerable(IEnumerable enumerable) + { + _enumerator = enumerable.GetEnumerator(); + } + + internal async Task<(bool HasValue, object Value)> PullAsync() + { + if (StreamMode != StreamMode.Pull || _asyncEnumerator == null) + throw new InvalidOperationException("Execution is not an asynchronous pull stream."); + + await _moveLock.WaitAsync(_cancellation.Token); + try + { + await WaitWhileHaltedAsync(); + _cancellation.Token.ThrowIfCancellationRequested(); + + var hasValue = await _asyncEnumerator.MoveNextAsync(); + return (hasValue, hasValue ? _asyncEnumerator.Current : null); + } + finally + { + _moveLock.Release(); + } + } + + internal async Task<(bool HasValue, object Value)> MoveNextAsync() + { + if (_enumerator == null) + throw new InvalidOperationException("Execution is not a synchronous stream."); + + await WaitWhileHaltedAsync(); + _cancellation.Token.ThrowIfCancellationRequested(); + + var hasValue = _enumerator.MoveNext(); + return (hasValue, hasValue ? _enumerator.Current : null); + } + + internal bool Halt() + { + lock (_stateLock) + { + if (Ended || !Pausable || _halted) + return false; + + _halted = true; + _resumeSignal = NewSignal(); + return true; + } + } + + internal bool Resume() + { + TaskCompletionSource resumeSignal; + + lock (_stateLock) + { + if (Ended || !Pausable || !_halted) + return false; + + _halted = false; + resumeSignal = _resumeSignal; + } + + resumeSignal.TrySetResult(true); + return true; + } + + /// + /// Asynchronously waits until a halted execution is resumed. + /// + public async Task WaitWhileHaltedAsync() + { + Task resumeTask; + + lock (_stateLock) + resumeTask = _resumeSignal.Task; + + await resumeTask; + _cancellation.Token.ThrowIfCancellationRequested(); + } + + internal async Task TerminateAsync() + { + TaskCompletionSource resumeSignal; + + lock (_stateLock) + { + if (Ended) + return; + + Ended = true; + _halted = false; + resumeSignal = _resumeSignal; + } + + resumeSignal.TrySetResult(true); + _cancellation.Cancel(); + + if (_asyncEnumerator != null) + await _asyncEnumerator.DisposeAsync(); + + (_enumerator as IDisposable)?.Dispose(); + } + + internal async Task EndAsync() + { + TaskCompletionSource resumeSignal; + + lock (_stateLock) + { + if (Ended) + return; + + Ended = true; + _halted = false; + resumeSignal = _resumeSignal; + } + + resumeSignal.TrySetResult(true); + + if (_asyncEnumerator != null) + await _asyncEnumerator.DisposeAsync(); + + (_enumerator as IDisposable)?.Dispose(); + } + + static TaskCompletionSource NewSignal() + => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + static TaskCompletionSource CompletedSignal() + { + var signal = NewSignal(); + signal.SetResult(true); + return signal; } } + + interface IAsyncEnumeratorAdapter + { + object Current { get; } + ValueTask MoveNextAsync(); + ValueTask DisposeAsync(); + } + + sealed class AsyncEnumeratorAdapter : IAsyncEnumeratorAdapter + { + readonly IAsyncEnumerator _enumerator; + + public object Current => _enumerator.Current; + + public AsyncEnumeratorAdapter(IAsyncEnumerable enumerable, CancellationToken cancellationToken) + { + _enumerator = enumerable.GetAsyncEnumerator(cancellationToken); + } + + public ValueTask MoveNextAsync() => _enumerator.MoveNextAsync(); + public ValueTask DisposeAsync() => _enumerator.DisposeAsync(); + } } diff --git a/Libraries/Esiur/Data/Types/ConstantDef.cs b/Libraries/Esiur/Data/Types/ConstantDef.cs index af003b1..17c72c7 100644 --- a/Libraries/Esiur/Data/Types/ConstantDef.cs +++ b/Libraries/Esiur/Data/Types/ConstantDef.cs @@ -14,7 +14,6 @@ public class ConstantDef : MemberDef { public object Value { get; set; } - public Map Annotations { get; set; } public Tru ValueType { get; set; } public FieldInfo FieldInfo { get; set; } @@ -128,7 +127,7 @@ public class ConstantDef : MemberDef - return new ConstantDef() + return DefinitionAttributeReader.Apply(ci, new ConstantDef() { Name = customName, Index = index, @@ -137,7 +136,7 @@ public class ConstantDef : MemberDef Value = value, FieldInfo = ci, Annotations = annotations, - }; + }); } diff --git a/Libraries/Esiur/Data/Types/DefinitionAttributeReader.cs b/Libraries/Esiur/Data/Types/DefinitionAttributeReader.cs new file mode 100644 index 0000000..a6ce8fe --- /dev/null +++ b/Libraries/Esiur/Data/Types/DefinitionAttributeReader.cs @@ -0,0 +1,74 @@ +using Esiur.Resource; +using System; +using System.Linq; +using System.Reflection; + +namespace Esiur.Data.Types; + +internal static class DefinitionAttributeReader +{ + internal static void Apply(Type source, TypeDef target) + { + target.Usage = source.GetCustomAttribute(true)?.Value; + target.Description = source.GetCustomAttribute(true)?.Value; + target.Example = source.GetCustomAttributes(true) + .Select(x => x.Value) + .FirstOrDefault(); + target.Category = source.GetCustomAttribute(true)?.Value; + target.Since = source.GetCustomAttribute(true)?.Value; + } + + internal static T Apply(MemberInfo source, T target) where T : MemberDef + { + target.Description = source.GetCustomAttribute(true)?.Value; + target.Usage = source.GetCustomAttribute(true)?.Value; + + var examples = source.GetCustomAttributes(true) + .Select(x => x.Value) + .ToList(); + target.Examples = examples.Count == 0 ? null : examples; + + var tags = source.GetCustomAttribute(true)?.Values; + target.Tags = tags == null || tags.Length == 0 ? null : tags.ToList(); + target.Unit = source.GetCustomAttribute(true)?.Value; + target.Minimum = source.GetCustomAttribute(true)?.Value; + target.Maximum = source.GetCustomAttribute(true)?.Value; + + var allowedValues = source.GetCustomAttributes(true) + .Select(x => x.Value) + .ToList(); + target.AllowedValues = allowedValues.Count == 0 ? null : allowedValues; + + target.Pattern = source.GetCustomAttribute(true)?.Value; + target.Format = source.GetCustomAttribute(true)?.Value; + + var preconditions = source.GetCustomAttributes(true) + .Select(x => x.Value) + .ToList(); + target.Preconditions = preconditions.Count == 0 ? null : preconditions; + + var postconditions = source.GetCustomAttributes(true) + .Select(x => x.Value) + .ToList(); + target.Postconditions = postconditions.Count == 0 ? null : postconditions; + + target.Effects = source.GetCustomAttribute(true)?.Value + ?? OperationEffects.None; + + var warnings = source.GetCustomAttributes(true) + .Select(x => x.Value) + .ToList(); + target.Warnings = warnings.Count == 0 ? null : warnings; + + var relatedMembers = source.GetCustomAttribute(true)?.Indexes; + target.RelatedMembers = relatedMembers == null || relatedMembers.Length == 0 + ? null + : relatedMembers.ToList(); + + var obsolete = source.GetCustomAttribute(true); + target.Deprecated = obsolete != null; + target.DeprecationMessage = obsolete?.Message; + + return target; + } +} diff --git a/Libraries/Esiur/Data/Types/EventDef.cs b/Libraries/Esiur/Data/Types/EventDef.cs index 2710e72..537c56f 100644 --- a/Libraries/Esiur/Data/Types/EventDef.cs +++ b/Libraries/Esiur/Data/Types/EventDef.cs @@ -15,12 +15,6 @@ namespace Esiur.Data.Types; public class EventDef : MemberDef { - public Map Annotations - { - get; - set; - } - public override string ToString() { return $"{Name}: {ArgumentType}"; @@ -35,7 +29,9 @@ public class EventDef : MemberDef set => AutoDelivered = !value; } - public bool Deprecated { get; set; } + public bool Historical { get; set; } + + public OrderingControl OrderingControl { get; set; } public EventInfo EventInfo { get; set; } @@ -166,7 +162,9 @@ public class EventDef : MemberDef throw new Exception($"Unsupported type `{argType}` in event `{type.Name}.{ei.Name}`"); var annotationAttrs = ei.GetCustomAttributes(true); - var autoDeliveredAttr = ei.GetCustomAttribute(true); + var autoDeliveryAttr = ei.GetCustomAttribute(true); + var historicalAttr = ei.GetCustomAttribute(true); + var orderingAttr = ei.GetCustomAttribute(true); //evtType.Nullable = new NullabilityInfoContext().Create(ei).ReadState is NullabilityState.Nullable; @@ -209,7 +207,7 @@ public class EventDef : MemberDef } - return new EventDef() + return DefinitionAttributeReader.Apply(ei, new EventDef() { Name = name, ArgumentType = evtType, @@ -217,8 +215,10 @@ public class EventDef : MemberDef Inherited = ei.DeclaringType != type, Annotations = annotations, EventInfo = ei, - AutoDelivered = autoDeliveredAttr != null - }; + AutoDelivered = autoDeliveryAttr != null, + Historical = historicalAttr != null, + OrderingControl = orderingAttr?.Control ?? OrderingControl.Strict, + }); } } diff --git a/Libraries/Esiur/Data/Types/FunctionDef.cs b/Libraries/Esiur/Data/Types/FunctionDef.cs index cdeda32..b3f8de3 100644 --- a/Libraries/Esiur/Data/Types/FunctionDef.cs +++ b/Libraries/Esiur/Data/Types/FunctionDef.cs @@ -15,12 +15,6 @@ namespace Esiur.Data.Types; public class FunctionDef : MemberDef { - public Map Annotations - { - get; - set; - } - //public bool IsVoid //{ // get; @@ -33,7 +27,7 @@ public class FunctionDef : MemberDef public bool ReadOnly { get; set; } public bool Idempotent { get; set; } public bool Cancellable { get; set; } - public bool Deprecated { get; set; } + public bool Pausable { get; set; } //public FunctionDefFlags Flags { get; set; } public StreamMode StreamMode { get; set; } @@ -175,28 +169,61 @@ public class FunctionDef : MemberDef { var genericRtType = mi.ReturnType.IsGenericType ? mi.ReturnType.GetGenericTypeDefinition() : null; - + var streamAttribute = mi.GetCustomAttribute(true); + var streamMode = StreamMode.None; + var pausable = false; Tru rtType; - if (genericRtType == typeof(AsyncReply<>)) + if (streamAttribute != null && + genericRtType != typeof(AsyncReply<>) && + genericRtType != typeof(AsyncStreamReply<>)) + throw new Exception($"Method `{type.Name}.{mi.Name}` uses StreamAttribute and must return AsyncReply or AsyncStreamReply."); + + if (genericRtType == typeof(IAsyncEnumerable<>)) { + streamMode = StreamMode.Pull; rtType = Tru.FromType(mi.ReturnType.GetGenericArguments()[0], warehouse); } - else if (genericRtType == typeof(Task<>)) + else if (genericRtType == typeof(IEnumerable<>)) { + streamMode = StreamMode.Push; rtType = Tru.FromType(mi.ReturnType.GetGenericArguments()[0], warehouse); } - else if (genericRtType == typeof(IEnumerable<>) || genericRtType == typeof(IAsyncEnumerable<>)) + else if (genericRtType == typeof(AsyncStreamReply<>)) + { + if (streamAttribute == null) + throw new Exception($"Method `{type.Name}.{mi.Name}` returning AsyncStreamReply must declare StreamAttribute."); + + streamMode = streamAttribute.Mode; + rtType = Tru.FromType(mi.ReturnType.GetGenericArguments()[0], warehouse); + } + else if (streamAttribute != null) + { + streamMode = streamAttribute.Mode; + rtType = Tru.FromType(mi.ReturnType.GetGenericArguments()[0], warehouse); + } + else if (genericRtType == typeof(AsyncReply<>) || genericRtType == typeof(Task<>)) { - // get export rtType = Tru.FromType(mi.ReturnType.GetGenericArguments()[0], warehouse); } else { - if (mi.ReturnType == typeof(Task)) - rtType = Tru.FromType(null, warehouse); - else - rtType = Tru.FromType(mi.ReturnType, warehouse); + rtType = mi.ReturnType == typeof(Task) + ? Tru.FromType(null, warehouse) + : Tru.FromType(mi.ReturnType, warehouse); + } + + if (streamAttribute != null) + { + if (streamAttribute.Mode != StreamMode.Push && streamAttribute.Mode != StreamMode.Pull) + throw new Exception($"Stream method `{type.Name}.{mi.Name}` must use Push or Pull mode."); + + if (streamMode != streamAttribute.Mode) + throw new Exception($"Stream mode `{streamAttribute.Mode}` conflicts with return type `{mi.ReturnType}` in method `{type.Name}.{mi.Name}`."); + + pausable = streamAttribute.Pausable; + if (pausable && streamMode != StreamMode.Push) + throw new Exception($"Only push stream method `{type.Name}.{mi.Name}` can be pausable."); } if (rtType == null) @@ -237,7 +264,12 @@ public class FunctionDef : MemberDef //var rtFlags = rtNullableAttr?.Flags?.ToList() ?? new List(); //var rtFlags = ((byte[])rtNullableAttr?.NullableFlags ?? new byte[0]).ToList(); - if (rtNullableAttrFlags.Count > 0 && genericRtType == typeof(AsyncReply<>)) + if (rtNullableAttrFlags.Count > 0 && + (genericRtType == typeof(AsyncReply<>) || + genericRtType == typeof(Task<>) || + genericRtType == typeof(IEnumerable<>) || + genericRtType == typeof(IAsyncEnumerable<>) || + genericRtType == typeof(AsyncStreamReply<>))) rtNullableAttrFlags.RemoveAt(0); if (rtNullableContextAttrFlag == 2) @@ -333,7 +365,7 @@ public class FunctionDef : MemberDef } - return new FunctionDef() + return DefinitionAttributeReader.Apply(mi, new FunctionDef() { Name = name, Index = index, @@ -342,8 +374,14 @@ public class FunctionDef : MemberDef ReturnType = rtType, Arguments = arguments, MethodInfo = mi, - Annotations = annotations - }; + Annotations = annotations, + ReadOnly = mi.GetCustomAttribute(true) != null, + Idempotent = mi.GetCustomAttribute(true) != null, + Cancellable = mi.GetCustomAttribute(true) != null, + RatePolicyName = mi.GetCustomAttribute(true)?.PolicyName, + StreamMode = streamMode, + Pausable = pausable, + }); } diff --git a/Libraries/Esiur/Data/Types/FunctionDefFlags.cs b/Libraries/Esiur/Data/Types/FunctionDefFlags.cs index 20a57e7..e9e9ddc 100644 --- a/Libraries/Esiur/Data/Types/FunctionDefFlags.cs +++ b/Libraries/Esiur/Data/Types/FunctionDefFlags.cs @@ -15,5 +15,6 @@ namespace Esiur.Data.Types ReadOnly = 0x08, Idempotent = 0x10, Cancellable = 0x20, + Pausable = 0x40, } -} \ No newline at end of file +} diff --git a/Libraries/Esiur/Data/Types/LocalTypeDef.cs b/Libraries/Esiur/Data/Types/LocalTypeDef.cs index 663cfd0..5b1b2eb 100644 --- a/Libraries/Esiur/Data/Types/LocalTypeDef.cs +++ b/Libraries/Esiur/Data/Types/LocalTypeDef.cs @@ -61,6 +61,9 @@ public class LocalTypeDef:TypeDef if (genericType == typeof(List<>) || genericType == typeof(PropertyContext<>) || genericType == typeof(AsyncReply<>) + || genericType == typeof(Task<>) + || genericType == typeof(IEnumerable<>) + || genericType == typeof(IAsyncEnumerable<>) || genericType == typeof(ResourceLink<>)) { return GetDistributedTypes(genericTypeArgs[0]); @@ -273,6 +276,8 @@ public class LocalTypeDef:TypeDef _typeName = GetTypeName(type); + DefinitionAttributeReader.Apply(type, this); + warehouse.TryRegisterLocalTypeDef(this); diff --git a/Libraries/Esiur/Data/Types/MemberDef.cs b/Libraries/Esiur/Data/Types/MemberDef.cs index a93daa2..8bdcb7a 100644 --- a/Libraries/Esiur/Data/Types/MemberDef.cs +++ b/Libraries/Esiur/Data/Types/MemberDef.cs @@ -16,6 +16,14 @@ public class MemberDef public bool Inherited { get; set; } + public bool Deprecated { get; set; } + + /// + /// Name of the server-side Warehouse rate policy applied to this member. + /// This is local execution metadata and is not sent to remote clients. + /// + public string? RatePolicyName { get; set; } + public TypeDef Definition { get; set; } = null!; // Human-readable metadata @@ -54,6 +62,8 @@ public class MemberDef // Compatibility guidance public string? DeprecationMessage { get; set; } + public Map? Annotations { get; set; } + public string Fullname => Definition is null || string.IsNullOrEmpty(Definition.Name) ? Name diff --git a/Libraries/Esiur/Data/Types/MemberDefField.cs b/Libraries/Esiur/Data/Types/MemberDefField.cs index 0fb550f..ac46d67 100644 --- a/Libraries/Esiur/Data/Types/MemberDefField.cs +++ b/Libraries/Esiur/Data/Types/MemberDefField.cs @@ -34,5 +34,8 @@ namespace Esiur.Data.Types // Compatibility guidance DeprecationMessage = 0x2F, // string + + // Extensible application metadata + Annotations = 0x30, // Map } } diff --git a/Libraries/Esiur/Data/Types/MemberDefInfo.cs b/Libraries/Esiur/Data/Types/MemberDefInfo.cs index e075640..c57047b 100644 --- a/Libraries/Esiur/Data/Types/MemberDefInfo.cs +++ b/Libraries/Esiur/Data/Types/MemberDefInfo.cs @@ -68,4 +68,7 @@ public class MemberDefInfo : IndexedStructure [Index((int)MemberDefField.DeprecationMessage)] public string? DeprecationMessage { get; set; } + + [Index((int)MemberDefField.Annotations)] + public Map? Annotations { get; set; } } diff --git a/Libraries/Esiur/Data/Types/PropertyDef.cs b/Libraries/Esiur/Data/Types/PropertyDef.cs index feabe22..4512396 100644 --- a/Libraries/Esiur/Data/Types/PropertyDef.cs +++ b/Libraries/Esiur/Data/Types/PropertyDef.cs @@ -14,11 +14,6 @@ namespace Esiur.Data.Types; public class PropertyDef : MemberDef { - public Map Annotations { get; set; } - - - - public PropertyInfo PropertyInfo { get; @@ -52,10 +47,15 @@ public class PropertyDef : MemberDef } public bool Constant { get; set; } + public bool Volatile { get; set; } //public bool IsNullable { get; set; } public bool Historical { get; set; } + public OrderingControl OrderingControl { get; set; } + + public object? DefaultValue { get; set; } + public bool HasHistory { get => Historical; @@ -258,6 +258,10 @@ public class PropertyDef : MemberDef var annotationAttrs = pi.GetCustomAttributes(true); var historicalAttr = pi.GetCustomAttribute(true); + var readOnlyAttr = pi.GetCustomAttribute(true); + var volatileAttr = pi.GetCustomAttribute(true); + var orderingAttr = pi.GetCustomAttribute(true); + var defaultValueAttr = pi.GetCustomAttribute(true); //var nullabilityContext = new NullabilityInfoContext(); //propType.Nullable = nullabilityContext.Create(pi).ReadState is NullabilityState.Nullable; @@ -306,16 +310,21 @@ public class PropertyDef : MemberDef } - return new PropertyDef() + return DefinitionAttributeReader.Apply(pi, new PropertyDef() { Name = name, Index = index, Inherited = pi.DeclaringType != type, ValueType = propType, PropertyInfo = pi, - Historical = historicalAttr == null, + RatePolicyName = pi.GetCustomAttribute(true)?.PolicyName, + ReadOnly = readOnlyAttr != null, + Volatile = volatileAttr != null, + Historical = historicalAttr != null, + OrderingControl = orderingAttr?.Control ?? OrderingControl.Strict, + DefaultValue = defaultValueAttr?.Value, Annotations = annotations, - }; + }); } @@ -325,7 +334,7 @@ public class PropertyDef : MemberDef PropertyPermission permission, TypeDef typeDef) { var definition = MakePropertyDef(warehouse, type, pi, name, index, typeDef); - definition.Permission = permission; + definition.ReadOnly = definition.ReadOnly || permission == PropertyPermission.Read; return definition; } diff --git a/Libraries/Esiur/Data/Types/RemoteTypeDef.cs b/Libraries/Esiur/Data/Types/RemoteTypeDef.cs index 6e6fc5a..d70c69d 100644 --- a/Libraries/Esiur/Data/Types/RemoteTypeDef.cs +++ b/Libraries/Esiur/Data/Types/RemoteTypeDef.cs @@ -194,6 +194,11 @@ public class RemoteTypeDef:TypeDef : $"{info.Namespace}.{info.Name}"; definition._typeDefKind = info.Kind; definition._version = info.Version; + definition.Usage = info.Usage; + definition.Description = info.Description; + definition.Example = info.Example; + definition.Category = info.Category; + definition.Since = info.Since; definition.Annotations = info.Annotations; definition._properties.Clear(); @@ -218,6 +223,7 @@ public class RemoteTypeDef:TypeDef member.Index = info.Index; member.Name = info.Name; member.Inherited = (info.Flags & (byte)MemberDefFlags.Inherited) != 0; + member.Deprecated = (info.Flags & (byte)MemberDefFlags.Deprecated) != 0; member.Description = info.Description; member.Usage = info.Usage; member.Examples = info.Examples; @@ -234,6 +240,7 @@ public class RemoteTypeDef:TypeDef member.Warnings = info.Warnings; member.RelatedMembers = info.RelatedMembers; member.DeprecationMessage = info.DeprecationMessage; + member.Annotations = info.Annotations; return member; } @@ -245,7 +252,10 @@ public class RemoteTypeDef:TypeDef ValueType = info.ValueType, ReadOnly = flags.HasFlag(PropertyDefFlags.ReadOnly), Constant = flags.HasFlag(PropertyDefFlags.Constant), + Volatile = flags.HasFlag(PropertyDefFlags.Volatile), Historical = flags.HasFlag(PropertyDefFlags.Historical) || info.HistoryControl != 0, + OrderingControl = info.OrderingControl, + DefaultValue = info.DefaultValue, }); } @@ -260,7 +270,7 @@ public class RemoteTypeDef:TypeDef ReadOnly = flags.HasFlag(FunctionDefFlags.ReadOnly), Idempotent = flags.HasFlag(FunctionDefFlags.Idempotent), Cancellable = flags.HasFlag(FunctionDefFlags.Cancellable), - Deprecated = flags.HasFlag(FunctionDefFlags.Deprecated), + Pausable = flags.HasFlag(FunctionDefFlags.Pausable), Arguments = info.Arguments?.Select(ToArgument).ToArray() ?? Array.Empty(), }); } @@ -275,6 +285,7 @@ public class RemoteTypeDef:TypeDef Optional = flags.HasFlag(ArgumentDefFlags.Optional), Type = info.ValueType, DefaultValue = info.DefaultValue, + Annotations = info.Annotations, }; } @@ -285,7 +296,8 @@ public class RemoteTypeDef:TypeDef { ArgumentType = info.ArgumentType, AutoDelivered = flags.HasFlag(EventDefFlags.AutoDelivered), - Deprecated = flags.HasFlag(EventDefFlags.Deprecated), + Historical = flags.HasFlag(EventDefFlags.Historical) || info.HistoryControl != 0, + OrderingControl = info.OrderingControl, }); } diff --git a/Libraries/Esiur/Data/Types/TypeDef.cs b/Libraries/Esiur/Data/Types/TypeDef.cs index 6e5b75c..d23e1c6 100644 --- a/Libraries/Esiur/Data/Types/TypeDef.cs +++ b/Libraries/Esiur/Data/Types/TypeDef.cs @@ -23,6 +23,16 @@ public class TypeDef public Map Annotations { get; set; } + public string? Usage { get; set; } + + public string? Description { get; set; } + + public object? Example { get; set; } + + public string? Category { get; set; } + + public string? Since { get; set; } + protected string _typeName; protected List _functions = new List(); protected List _events = new List(); diff --git a/Libraries/Esiur/Data/Types/TypeDefInfo.cs b/Libraries/Esiur/Data/Types/TypeDefInfo.cs index 8a82bc2..5d6ffe5 100644 --- a/Libraries/Esiur/Data/Types/TypeDefInfo.cs +++ b/Libraries/Esiur/Data/Types/TypeDefInfo.cs @@ -66,6 +66,11 @@ public class TypeDefInfo : IndexedStructure Name = definition.Name, Kind = definition.Kind, Parent = definition.ParentTypeId, + Usage = definition.Usage, + Description = definition.Description, + Example = definition.Example, + Category = definition.Category, + Since = definition.Since, Annotations = definition.Annotations, Properties = definition.Properties.Select(FromProperty).ToList(), Functions = definition.Functions.Select(FromFunction).ToList(), @@ -94,21 +99,26 @@ public class TypeDefInfo : IndexedStructure target.Warnings = source.Warnings; target.RelatedMembers = source.RelatedMembers; target.DeprecationMessage = source.DeprecationMessage; + target.Annotations = source.Annotations; return target; } private static PropertyDefInfo FromProperty(PropertyDef source) { var flags = source.Inherited ? PropertyDefFlags.Inherited : PropertyDefFlags.None; + if (source.Deprecated) flags |= PropertyDefFlags.Deprecated; if (source.ReadOnly) flags |= PropertyDefFlags.ReadOnly; if (source.Constant) flags |= PropertyDefFlags.Constant; + if (source.Volatile) flags |= PropertyDefFlags.Volatile; if (source.Historical) flags |= PropertyDefFlags.Historical; return CopyMember(source, new PropertyDefInfo { Flags = (byte)flags, ValueType = source.ValueType, + OrderingControl = source.OrderingControl, HistoryControl = source.Historical ? (byte)1 : (byte)0, + DefaultValue = source.DefaultValue, }); } @@ -120,6 +130,7 @@ public class TypeDefInfo : IndexedStructure if (source.ReadOnly) flags |= FunctionDefFlags.ReadOnly; if (source.Idempotent) flags |= FunctionDefFlags.Idempotent; if (source.Cancellable) flags |= FunctionDefFlags.Cancellable; + if (source.Pausable) flags |= FunctionDefFlags.Pausable; return CopyMember(source, new FunctionDefInfo { @@ -139,6 +150,7 @@ public class TypeDefInfo : IndexedStructure Flags = source.Optional ? (byte)ArgumentDefFlags.Optional : (byte)ArgumentDefFlags.None, ValueType = source.Type, DefaultValue = source.DefaultValue, + Annotations = source.Annotations, }; } @@ -147,17 +159,21 @@ public class TypeDefInfo : IndexedStructure var flags = source.Inherited ? EventDefFlags.Inherited : EventDefFlags.None; if (source.Deprecated) flags |= EventDefFlags.Deprecated; if (source.AutoDelivered) flags |= EventDefFlags.AutoDelivered; + if (source.Historical) flags |= EventDefFlags.Historical; return CopyMember(source, new EventDefInfo { Flags = (byte)flags, ArgumentType = source.ArgumentType, + OrderingControl = source.OrderingControl, + HistoryControl = source.Historical ? (byte)1 : (byte)0, }); } private static ConstantDefInfo FromConstant(ConstantDef source) { var flags = source.Inherited ? ConstantDefFlags.Inherited : ConstantDefFlags.None; + if (source.Deprecated) flags |= ConstantDefFlags.Deprecated; return CopyMember(source, new ConstantDefInfo { Flags = (byte)flags, diff --git a/Libraries/Esiur/Net/NetworkServer.cs b/Libraries/Esiur/Net/NetworkServer.cs index 27baf6c..09123a0 100644 --- a/Libraries/Esiur/Net/NetworkServer.cs +++ b/Libraries/Esiur/Net/NetworkServer.cs @@ -103,7 +103,7 @@ public abstract class NetworkServer : IDestructible where TConnecti if (s == null) { - Global.Log("NetworkServer", LogType.Error, "sock == null"); + //Global.Log("NetworkServer", LogType.Error, "sock == null"); return; } diff --git a/Libraries/Esiur/Net/Packets/EpPacketRequest.cs b/Libraries/Esiur/Net/Packets/EpPacketRequest.cs index 516d335..1e01cf4 100644 --- a/Libraries/Esiur/Net/Packets/EpPacketRequest.cs +++ b/Libraries/Esiur/Net/Packets/EpPacketRequest.cs @@ -37,6 +37,7 @@ namespace Esiur.Net.Packets PullStream = 0x1C, TerminateExecution = 0x1D, HaltExecution = 0x1E, + ResumeExecution = 0x1F, } } diff --git a/Libraries/Esiur/Protocol/EpConnection.cs b/Libraries/Esiur/Protocol/EpConnection.cs index 58ec131..25961aa 100644 --- a/Libraries/Esiur/Protocol/EpConnection.cs +++ b/Libraries/Esiur/Protocol/EpConnection.cs @@ -454,6 +454,7 @@ public partial class EpConnection : NetworkConnection, IStore public override void Destroy() { + TerminateInvocations(); UnsubscribeAll(); this.OnReady = null; this.OnError = null; @@ -477,7 +478,9 @@ public partial class EpConnection : NetworkConnection, IStore { offset += (uint)rt; - if (_packet.Tdu == null) + if (_packet.Tdu == null && + _packet.Method != EpPacketMethod.Reply && + _packet.Method != EpPacketMethod.Extension) return offset; #if VERBOSE @@ -514,6 +517,9 @@ public partial class EpConnection : NetworkConnection, IStore } else if (_packet.Method == EpPacketMethod.Request) { + if (IsRateControlBlocked) + return offset; + var dt = _packet.Tdu.Value; switch (_packet.Request) @@ -582,37 +588,58 @@ public partial class EpConnection : NetworkConnection, IStore case EpPacketRequest.StaticCall: EpRequestStaticCall(_packet.CallbackId, dt); break; + case EpPacketRequest.PullStream: + EpRequestPullStream(_packet.CallbackId, dt); + break; + case EpPacketRequest.TerminateExecution: + EpRequestTerminateExecution(_packet.CallbackId, dt); + break; + case EpPacketRequest.HaltExecution: + EpRequestHaltExecution(_packet.CallbackId, dt); + break; + case EpPacketRequest.ResumeExecution: + EpRequestResumeExecution(_packet.CallbackId, dt); + break; } } else if (_packet.Method == EpPacketMethod.Reply) { - var dt = _packet.Tdu.Value; + var dt = _packet.Tdu; switch (_packet.Reply) { case EpPacketReply.Completed: EpReplyCompleted(_packet.CallbackId, dt); break; + case EpPacketReply.Stream: + EpReplyStream(_packet.CallbackId); + break; case EpPacketReply.Propagated: - EpReplyPropagated(_packet.CallbackId, dt); + if (dt.HasValue) + EpReplyPropagated(_packet.CallbackId, dt.Value); break; case EpPacketReply.PermissionError: - EpReplyError(_packet.CallbackId, dt, ErrorType.Management); + if (dt.HasValue) + EpReplyError(_packet.CallbackId, dt.Value, ErrorType.Management); break; case EpPacketReply.ExecutionError: - EpReplyError(_packet.CallbackId, dt, ErrorType.Exception); + if (dt.HasValue) + EpReplyError(_packet.CallbackId, dt.Value, ErrorType.Exception); break; case EpPacketReply.Progress: - EpReplyProgress(_packet.CallbackId, dt); + if (dt.HasValue) + EpReplyProgress(_packet.CallbackId, dt.Value); break; case EpPacketReply.Chunk: - EpReplyChunk(_packet.CallbackId, dt); + if (dt.HasValue) + EpReplyChunk(_packet.CallbackId, dt.Value); break; case EpPacketReply.Warning: - EpReplyWarning(_packet.Extension, dt); + if (dt.HasValue) + EpReplyWarning(_packet.CallbackId, dt.Value); break; } @@ -2102,6 +2129,7 @@ public partial class EpConnection : NetworkConnection, IStore protected override void Disconnected() { // clean up + TerminateInvocations(); _authenticated = false; Status = EpConnectionStatus.Closed; diff --git a/Libraries/Esiur/Protocol/EpConnectionProtocol.cs b/Libraries/Esiur/Protocol/EpConnectionProtocol.cs index ac020a7..025e552 100644 --- a/Libraries/Esiur/Protocol/EpConnectionProtocol.cs +++ b/Libraries/Esiur/Protocol/EpConnectionProtocol.cs @@ -31,6 +31,7 @@ using Esiur.Net.Packets; using Esiur.Resource; using Esiur.Security.Authority; using Esiur.Security.Permissions; +using Esiur.Security.RateLimiting; using Microsoft.CodeAnalysis.CSharp.Syntax; using System; using System.Collections; @@ -47,6 +48,140 @@ namespace Esiur.Protocol; partial class EpConnection { + readonly object _rateControlDenialsLock = new object(); + DateTime _rateControlDenialWindowStarted; + int _rateControlDenials; + int _rateControlBlocked; + + internal bool IsRateControlBlocked => Volatile.Read(ref _rateControlBlocked) != 0; + + bool TryApplyRateControl( + MemberDef member, + IResource? resource, + ActionType action, + uint callback, + out TimeSpan delay) + { + delay = TimeSpan.Zero; + + if (string.IsNullOrWhiteSpace(member.RatePolicyName)) + return true; + + if (IsRateControlBlocked) + return false; + + var warehouse = Instance.Warehouse; + var policy = warehouse.TryGetRatePolicy(member.RatePolicyName); + + if (policy == null) + { + DenyRateControlledRequest( + callback, + $"Rate policy `{member.RatePolicyName}` is not registered."); + return false; + } + + try + { + var context = new RateControlContext( + warehouse, + this, + _session, + resource, + member, + action); + + if (policy.Applicable(context) == Ruling.Denied) + { + DenyRateControlledRequest( + callback, + $"Rate policy `{member.RatePolicyName}` denied `{member.Fullname}`."); + return false; + } + + delay = context.Delay > TimeSpan.Zero ? context.Delay : TimeSpan.Zero; + return true; + } + catch (Exception exception) + { + DenyRateControlledRequest( + callback, + $"Rate policy `{member.RatePolicyName}` failed: {exception.Message}"); + return false; + } + } + + void DenyRateControlledRequest(uint callback, string message) + { + var configuration = Instance.Warehouse.Configuration.RateControl; + var now = DateTime.UtcNow; + var shouldBlock = false; + + lock (_rateControlDenialsLock) + { + var window = configuration.DenialWindow > TimeSpan.Zero + ? configuration.DenialWindow + : TimeSpan.FromMinutes(1); + + if (_rateControlDenialWindowStarted == default || + now - _rateControlDenialWindowStarted > window) + { + _rateControlDenialWindowStarted = now; + _rateControlDenials = 0; + } + + _rateControlDenials++; + shouldBlock = configuration.DenialsBeforeConnectionBlock > 0 && + _rateControlDenials >= configuration.DenialsBeforeConnectionBlock && + Interlocked.Exchange(ref _rateControlBlocked, 1) == 0; + } + + SendError( + ErrorType.Management, + callback, + (ushort)ExceptionCode.RateLimitExceeded, + message); + + if (shouldBlock) + _ = CloseRateControlledConnectionAsync(configuration.ConnectionBlockDelay); + } + + async Task CloseRateControlledConnectionAsync(TimeSpan delay) + { + if (delay > TimeSpan.Zero) + await Task.Delay(delay).ConfigureAwait(false); + + Close(); + } + + void ExecuteRateControlled(uint callback, TimeSpan delay, Action action) + { + if (delay <= TimeSpan.Zero) + { + action(); + return; + } + + _ = ExecuteRateControlledAsync(callback, delay, action); + } + + async Task ExecuteRateControlledAsync(uint callback, TimeSpan delay, Action action) + { + await Task.Delay(delay).ConfigureAwait(false); + + if (!IsConnected || IsRateControlBlocked) + return; + + try + { + action(); + } + catch (Exception exception) + { + SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.GeneralFailure, exception.Message); + } + } + KeyList _neededTypeDefs = new KeyList(); KeyList _cachedTypeDefs = new KeyList(); KeyList> _typeDefRequests = new KeyList>(); @@ -94,6 +229,9 @@ partial class EpConnection KeyList _requests = new KeyList(); + readonly object _invocationsLock = new object(); + readonly Dictionary _invocations = new Dictionary(); + volatile int _callbackCounter = 0; Dictionary> _subscriptions = new Dictionary>(); @@ -125,18 +263,25 @@ partial class EpConnection //callbackCounter++; // avoid thread racing _requests.Add(c, reply); + SendRequestPacket(action, c, args); + return reply; + } + + void SendRequestPacket(EpPacketRequest action, uint callbackId, object[] args) + { + if (args.Length == 0) { var bl = new BinaryList(); bl.AddUInt8((byte)(0x40 | (byte)action)) - .AddUInt32(c); + .AddUInt32(callbackId); Send(bl.ToArray()); } if (args.Length == 1) { var bl = new BinaryList(); bl.AddUInt8((byte)(0x60 | (byte)action)) - .AddUInt32(c) + .AddUInt32(callbackId) .AddUInt8Array(Codec.Compose(args[0], this.Instance?.Warehouse ?? _serverWarehouse, this)); Send(bl.ToArray()); } @@ -144,11 +289,39 @@ partial class EpConnection { var bl = new BinaryList(); bl.AddUInt8((byte)(0x60 | (byte)action)) - .AddUInt32(c) + .AddUInt32(callbackId) .AddUInt8Array(Codec.Compose(args, this.Instance?.Warehouse ?? _serverWarehouse, this)); Send(bl.ToArray()); } + } + AsyncStreamReply SendStreamRequest(StreamMode streamMode, EpPacketRequest action, params object[] args) + { + var callbackId = (uint)Interlocked.Increment(ref _callbackCounter); + var reply = new AsyncStreamReply( + streamMode, + () => SendRequest(EpPacketRequest.PullStream, callbackId), + () => SendRequest(EpPacketRequest.TerminateExecution, callbackId), + () => SendRequest(EpPacketRequest.HaltExecution, callbackId), + () => SendRequest(EpPacketRequest.ResumeExecution, callbackId)); + + _requests.Add(callbackId, reply); + SendRequestPacket(action, callbackId, args); + return reply; + } + + AsyncStreamReply SendStreamRequest(StreamMode streamMode, EpPacketRequest action, params object[] args) + { + var callbackId = (uint)Interlocked.Increment(ref _callbackCounter); + var reply = new AsyncStreamReply( + streamMode, + () => SendRequest(EpPacketRequest.PullStream, callbackId), + () => SendRequest(EpPacketRequest.TerminateExecution, callbackId), + () => SendRequest(EpPacketRequest.HaltExecution, callbackId), + () => SendRequest(EpPacketRequest.ResumeExecution, callbackId)); + + _requests.Add(callbackId, reply); + SendRequestPacket(action, callbackId, args); return reply; } @@ -304,6 +477,16 @@ partial class EpConnection return SendRequest(EpPacketRequest.StaticCall, typeId, index, parameters); } + public AsyncStreamReply StaticStreamCall(ulong typeId, byte index, object parameters, StreamMode streamMode) + { + return SendStreamRequest(streamMode, EpPacketRequest.StaticCall, typeId, index, parameters); + } + + public AsyncStreamReply StaticStreamCall(ulong typeId, byte index, object parameters, StreamMode streamMode) + { + return SendStreamRequest(streamMode, EpPacketRequest.StaticCall, typeId, index, parameters); + } + public AsyncReply Call(string procedureCall, params object[] parameters) { //var args = new Map(); @@ -324,6 +507,16 @@ partial class EpConnection return SendRequest(EpPacketRequest.InvokeFunction, instanceId, index, parameters); } + internal AsyncStreamReply SendStreamInvoke(uint instanceId, byte index, object parameters, StreamMode streamMode) + { + return SendStreamRequest(streamMode, EpPacketRequest.InvokeFunction, instanceId, index, parameters); + } + + internal AsyncStreamReply SendStreamInvoke(uint instanceId, byte index, object parameters, StreamMode streamMode) + { + return SendStreamRequest(streamMode, EpPacketRequest.InvokeFunction, instanceId, index, parameters); + } + internal AsyncReply SendSetProperty(uint instanceId, byte index, object value) { return SendRequest(EpPacketRequest.SetProperty, instanceId, index, value); @@ -383,7 +576,7 @@ partial class EpConnection SendReply(EpPacketReply.Chunk, callbackId, chunk); } - void EpReplyCompleted(uint callbackId, PlainTdu tdu) + void EpReplyCompleted(uint callbackId, PlainTdu? tdu) { var req = _requests.Take(callbackId); @@ -395,7 +588,19 @@ partial class EpConnection return; } - var pr = Codec.Parse(tdu, this, null); + if (req is AsyncStreamReply streamReply) + { + streamReply.TriggerStreamCompleted(); + return; + } + + if (tdu == null) + { + req.Trigger(null); + return; + } + + var pr = Codec.Parse(tdu.Value, this, null); if (pr is AsyncReply asyncReply) { @@ -421,6 +626,12 @@ partial class EpConnection //}).Error(req.TriggerError); } + void EpReplyStream(uint callbackId) + { + if (_requests[callbackId] is AsyncStreamReply streamReply) + streamReply.TriggerStreamStarted(); + } + void EpExtensionAction(byte actionId, PlainTdu? tdu) { // nothing is supported now @@ -1419,6 +1630,22 @@ partial class EpConnection void InvokeFunction(FunctionDef ft, uint callback, object arguments, EpPacketRequest actionType, object target = null) + { + if (!TryApplyRateControl( + ft, + target as IResource, + ActionType.Execute, + callback, + out var delay)) + return; + + ExecuteRateControlled( + callback, + delay, + () => InvokeFunctionCore(ft, callback, arguments, actionType, target)); + } + + void InvokeFunctionCore(FunctionDef ft, uint callback, object arguments, EpPacketRequest actionType, object target = null) { // cast arguments @@ -1559,59 +1786,44 @@ partial class EpConnection return; } - if (rt is IAsyncEnumerable) + if (ft.StreamMode != StreamMode.None) { - var enu = rt as IAsyncEnumerable; - var enumerator = enu.GetAsyncEnumerator(); - Task.Run(async () => + context ??= new InvocationContext(this, callback); + context.InitializeStream(ft.StreamMode, ft.Pausable); + + if (ft.StreamMode == StreamMode.Pull && context.SetAsyncEnumerable(rt)) { - try - { - while (await enumerator.MoveNextAsync()) + RegisterInvocation(callback, context); + SendReply(EpPacketReply.Stream, callback); + } + else if (ft.StreamMode == StreamMode.Push && rt is System.Collections.IEnumerable enumerable) + { + context.SetEnumerable(enumerable); + RegisterInvocation(callback, context); + SendReply(EpPacketReply.Stream, callback); + PumpEnumerable(callback, context); + } + else if (ft.StreamMode == StreamMode.Push && rt is AsyncReply streamSource) + { + RegisterInvocation(callback, context); + SendReply(EpPacketReply.Stream, callback); + + streamSource.Then(_ => CompleteInvocation(callback, context)) + .Error(ex => FailInvocation(callback, context, ex)) + .Progress((pt, pv, pm) => SendProgress(callback, pv, pm)) + .Chunk(value => { - var v = enumerator.Current; - SendChunk(callback, v); - } - - SendReply(EpPacketReply.Completed, callback); - - if (context != null) - context.Ended = true; - } - catch (Exception ex) - { - if (context != null) - context.Ended = true; - - var (code, msg) = SummerizeException(ex); - SendError(ErrorType.Exception, callback, code, msg); - } - }); - } - else if (rt is System.Collections.IEnumerable && !(rt is Array || rt is Map || rt is string)) - { - var enu = rt as System.Collections.IEnumerable; - - try - { - foreach (var v in enu) - SendChunk(callback, v); - - SendReply(EpPacketReply.Completed, callback); - - if (context != null) - context.Ended = true; - + if (!context.Ended) + context.Chunk(value); + }) + .Warning((level, message) => SendWarning(callback, level, message)); } - catch (Exception ex) + else { - if (context != null) - context.Ended = true; - - var (code, msg) = SummerizeException(ex); - SendError(ErrorType.Exception, callback, code, msg); + context.Ended = true; + SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.NotSupported, + $"Stream mode `{ft.StreamMode}` is not compatible with `{rt?.GetType()}`."); } - } else if (rt is Task) { @@ -1663,6 +1875,217 @@ partial class EpConnection } } + void RegisterInvocation(uint callback, InvocationContext context) + { + lock (_invocationsLock) + _invocations[callback] = context; + } + + InvocationContext GetInvocation(uint callback) + { + lock (_invocationsLock) + return _invocations.TryGetValue(callback, out var context) ? context : null; + } + + void TerminateInvocations() + { + InvocationContext[] contexts; + + lock (_invocationsLock) + { + contexts = _invocations.Values.ToArray(); + _invocations.Clear(); + } + + foreach (var context in contexts) + Task.Run(async () => + { + try + { + await context.TerminateAsync(); + } + catch + { + // The connection is already closing; disposal is best effort. + } + }); + } + + bool TakeInvocation(uint callback, InvocationContext expected, out InvocationContext context) + { + lock (_invocationsLock) + { + if (!_invocations.TryGetValue(callback, out context) || + (expected != null && !ReferenceEquals(expected, context))) + return false; + + _invocations.Remove(callback); + return true; + } + } + + async void CompleteInvocation(uint callback, InvocationContext expected) + { + if (!TakeInvocation(callback, expected, out var context)) + return; + + try + { + await context.EndAsync(); + SendReply(EpPacketReply.Completed, callback); + } + catch (Exception ex) + { + var (code, message) = SummerizeException(ex); + SendError(ErrorType.Exception, callback, code, message); + } + } + + async void FailInvocation(uint callback, InvocationContext expected, Exception exception) + { + if (!TakeInvocation(callback, expected, out var context)) + return; + + try + { + await context.TerminateAsync(); + } + catch + { + // Preserve the exception that ended the stream. + } + + var (code, message) = SummerizeException(exception); + SendError(ErrorType.Exception, callback, code, message); + } + + void PumpEnumerable(uint callback, InvocationContext context) + { + Task.Run(async () => + { + try + { + while (true) + { + var next = await context.MoveNextAsync(); + if (!next.HasValue) + { + CompleteInvocation(callback, context); + return; + } + + context.Chunk(next.Value); + } + } + catch (OperationCanceledException) + { + // TerminateExecution owns completion when it cancels the pump. + } + catch (Exception ex) + { + FailInvocation(callback, context, ex); + } + }); + } + + uint ParseExecutionCallback(PlainTdu tdu) + => Convert.ToUInt32(Codec.ParseSync(tdu, Instance.Warehouse)); + + void EpRequestPullStream(uint callback, PlainTdu tdu) + { + var executionCallback = ParseExecutionCallback(tdu); + var context = GetInvocation(executionCallback); + + if (context == null || context.StreamMode != StreamMode.Pull) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAllowed, + "The target execution is not an active pull stream."); + return; + } + + Task.Run(async () => + { + try + { + var next = await context.PullAsync(); + if (next.HasValue) + SendChunk(executionCallback, next.Value); + else + CompleteInvocation(executionCallback, context); + + SendReply(EpPacketReply.Completed, callback); + } + catch (OperationCanceledException) + { + SendReply(EpPacketReply.Completed, callback); + } + catch (Exception ex) + { + FailInvocation(executionCallback, context, ex); + var (code, message) = SummerizeException(ex); + SendError(ErrorType.Exception, callback, code, message); + } + }); + } + + void EpRequestTerminateExecution(uint callback, PlainTdu tdu) + { + var executionCallback = ParseExecutionCallback(tdu); + + if (!TakeInvocation(executionCallback, null, out var context)) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAllowed, + "The target execution is not active."); + return; + } + + Task.Run(async () => + { + try + { + await context.TerminateAsync(); + SendReply(EpPacketReply.Completed, executionCallback); + SendReply(EpPacketReply.Completed, callback); + } + catch (Exception ex) + { + var (code, message) = SummerizeException(ex); + SendError(ErrorType.Exception, executionCallback, code, message); + SendError(ErrorType.Exception, callback, code, message); + } + }); + } + + void EpRequestHaltExecution(uint callback, PlainTdu tdu) + { + var executionCallback = ParseExecutionCallback(tdu); + var context = GetInvocation(executionCallback); + + if (context == null || !context.Halt()) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAllowed, + "The target execution is not pausable or is already halted."); + return; + } + + SendReply(EpPacketReply.Completed, callback); + } + + void EpRequestResumeExecution(uint callback, PlainTdu tdu) + { + var executionCallback = ParseExecutionCallback(tdu); + var context = GetInvocation(executionCallback); + + if (context == null || !context.Resume()) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAllowed, + "The target execution is not pausable or is not halted."); + return; + } + + SendReply(EpPacketReply.Completed, callback); + } + void EpRequestSubscribe(uint callback, PlainTdu tdu) { @@ -1683,7 +2106,7 @@ partial class EpConnection var et = r.Instance.Definition.GetEventDefByIndex(index); - if (et != null) + if (et == null) { // et not found SendError(ErrorType.Management, callback, (ushort)ExceptionCode.MethodNotFound); @@ -1789,7 +2212,7 @@ partial class EpConnection var (offset, length, args) = DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, tdu.PayloadLength, Instance.Warehouse, 2); - var rid = (uint)args[0]; + var rid = Convert.ToUInt32(args[0]); var index = (byte)args[1]; // un hold the socket to send data immediately @@ -1806,13 +2229,27 @@ partial class EpConnection var pt = r.Instance.Definition.GetPropertyDefByIndex(index); - if (pt != null) + if (pt == null) { // property not found SendError(ErrorType.Management, callback, (ushort)ExceptionCode.PropertyNotFound); return; } + if (r.Instance.Applicable(_session, ActionType.SetProperty, pt, this) == Ruling.Denied) + { + SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.SetPropertyDenied); + return; + } + + if (!TryApplyRateControl( + pt, + r, + ActionType.SetProperty, + callback, + out var rateControlDelay)) + return; + if (r is IDynamicResource dynamicResource) { @@ -1823,25 +2260,27 @@ partial class EpConnection asyncReply.Then((value) => { // propagation - dynamicResource.SetResourcePropertyAsync(index, value).Then((x) => - { - SendReply(EpPacketReply.Completed, callback); - }).Error(x => - { - SendError(x.Type, callback, (ushort)x.Code, x.Message); - }); + ExecuteRateControlled(callback, rateControlDelay, () => + dynamicResource.SetResourcePropertyAsync(index, value).Then((x) => + { + SendReply(EpPacketReply.Completed, callback); + }).Error(x => + { + SendError(x.Type, callback, (ushort)x.Code, x.Message); + })); }); } else { // propagation - dynamicResource.SetResourcePropertyAsync(index, pr.Value).Then((x) => - { - SendReply(EpPacketReply.Completed, callback); - }).Error(x => - { - SendError(x.Type, callback, (ushort)x.Code, x.Message); - }); + ExecuteRateControlled(callback, rateControlDelay, () => + dynamicResource.SetResourcePropertyAsync(index, pr.Value).Then((x) => + { + SendReply(EpPacketReply.Completed, callback); + }).Error(x => + { + SendError(x.Type, callback, (ushort)x.Code, x.Message); + })); } }).Error(x => SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError)); ; @@ -1856,12 +2295,6 @@ partial class EpConnection return; } - if (r.Instance.Applicable(_session, ActionType.SetProperty, pt, this) == Ruling.Denied) - { - SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.SetPropertyDenied); - return; - } - if (!pi.CanWrite) { SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ReadOnlyProperty); @@ -1876,6 +2309,36 @@ partial class EpConnection if (pr.Value is AsyncReply asyncReply) { asyncReply.Then((value) => + { + ExecuteRateControlled(callback, rateControlDelay, () => + { + if (pi.PropertyType.IsGenericType && pi.PropertyType.GetGenericTypeDefinition() + == typeof(PropertyContext<>)) + { + value = Activator.CreateInstance(pi.PropertyType, this, value); + } + else + { + value = RuntimeCaster.Cast(value, pi.PropertyType); + } + + try + { + pi.SetValue(r, value); + SendReply(EpPacketReply.Completed, callback); + } + catch (Exception ex) + { + SendError(ErrorType.Exception, callback, 0, ex.Message); + } + }); + }); + } + else + { + var value = pr.Value; + + ExecuteRateControlled(callback, rateControlDelay, () => { if (pi.PropertyType.IsGenericType && pi.PropertyType.GetGenericTypeDefinition() == typeof(PropertyContext<>)) @@ -1884,7 +2347,6 @@ partial class EpConnection } else { - // cast new value type to property type value = RuntimeCaster.Cast(value, pi.PropertyType); } @@ -1899,32 +2361,6 @@ partial class EpConnection } }); } - else - { - var value = pr.Value; - - if (pi.PropertyType.IsGenericType && pi.PropertyType.GetGenericTypeDefinition() - == typeof(PropertyContext<>)) - { - value = Activator.CreateInstance(pi.PropertyType, this, value); - //value = new DistributedPropertyContext(this, value); - } - else - { - // cast new value type to property type - value = RuntimeCaster.Cast(value, pi.PropertyType); - } - - try - { - pi.SetValue(r, value); - SendReply(EpPacketReply.Completed, callback); - } - catch (Exception ex) - { - SendError(ErrorType.Exception, callback, 0, ex.Message); - } - } }).Error(x => SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError)); } }); diff --git a/Libraries/Esiur/Protocol/EpResource.cs b/Libraries/Esiur/Protocol/EpResource.cs index 4754b93..fb350c1 100644 --- a/Libraries/Esiur/Protocol/EpResource.cs +++ b/Libraries/Esiur/Protocol/EpResource.cs @@ -289,9 +289,33 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn throw new Exception("Function definition not found."); if (ft.IsStatic) - return _connection.StaticCall(Instance.Definition.Id, index, args); - else - return _connection.SendInvoke(_instanceId, index, args); + return ft.StreamMode == StreamMode.None + ? _connection.StaticCall(Instance.Definition.Id, index, args) + : _connection.StaticStreamCall(Instance.Definition.Id, index, args, ft.StreamMode); + + return ft.StreamMode == StreamMode.None + ? _connection.SendInvoke(_instanceId, index, args) + : _connection.SendStreamInvoke(_instanceId, index, args, ft.StreamMode); + } + + public AsyncStreamReply _InvokeStream(byte index, object args) + { + if (_status == ResourceStatus.Destroyed) + throw new Exception("Trying to access a destroyed object."); + + if (_status == ResourceStatus.Suspended) + throw new Exception("Trying to access a suspended object."); + + var function = Instance.Definition.GetFunctionDefByIndex(index) + ?? throw new Exception("Function definition not found."); + + if (function.StreamMode == StreamMode.None) + throw new Exception("Function is not a stream."); + + if (function.IsStatic) + return _connection.StaticStreamCall(Instance.Definition.Id, index, args, function.StreamMode); + + return _connection.SendStreamInvoke(_instanceId, index, args, function.StreamMode); } public AsyncReply Subscribe(EventDef et) @@ -634,7 +658,8 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn // this only happens if the programmer forgot to emit in property setter _properties[index] = value; reply.Trigger(null); - }); + }) + .Error(reply.TriggerError); return reply; @@ -655,4 +680,4 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn { Destroy(); } -} \ No newline at end of file +} diff --git a/Libraries/Esiur/Proxy/ResourceGenerator.cs b/Libraries/Esiur/Proxy/ResourceGenerator.cs index 77dbd95..cebc24b 100644 --- a/Libraries/Esiur/Proxy/ResourceGenerator.cs +++ b/Libraries/Esiur/Proxy/ResourceGenerator.cs @@ -113,7 +113,7 @@ namespace Esiur.Proxy if (!ci.HasHandle) { - code.AppendLine("public virtual AsyncReply Handle(ResourceOperation operation, IResourceContext context = null) => new AsyncReply(true);"); + code.AppendLine("public virtual AsyncReply Handle(ResourceOperation operation, IResourceContext? context = null) => new AsyncReply(true);"); } } @@ -359,4 +359,4 @@ namespace Esiur.Proxy public bool HasHandle; } } -} \ No newline at end of file +} diff --git a/Libraries/Esiur/Proxy/TypeDefGenerator.cs b/Libraries/Esiur/Proxy/TypeDefGenerator.cs index 8f020b8..97ce3c3 100644 --- a/Libraries/Esiur/Proxy/TypeDefGenerator.cs +++ b/Libraries/Esiur/Proxy/TypeDefGenerator.cs @@ -322,7 +322,7 @@ public static class TypeDefGenerator var rt = new StringBuilder(); - rt.AppendLine("using System;\r\nusing Esiur.Resource;\r\nusing Esiur.Core;\r\nusing Esiur.Data;\r\nusing Esiur.Protocol;"); + rt.AppendLine("using System;\r\nusing Esiur.Resource;\r\nusing Esiur.Core;\r\nusing Esiur.Data;\r\nusing Esiur.Data.Types;\r\nusing Esiur.Protocol;"); rt.AppendLine("#nullable enable"); rt.AppendLine($"namespace {nameSpace} {{"); @@ -355,6 +355,8 @@ public static class TypeDefGenerator continue; var rtTypeName = GetTypeName(f.ReturnType, typeDefs); + var isStream = f.StreamMode != StreamMode.None; + var replyType = isStream ? "AsyncStreamReply" : "AsyncReply"; var positionalArgs = f.Arguments.Where((x) => !x.Optional).ToArray(); var optionalArgs = f.Arguments.Where((x) => x.Optional).ToArray(); @@ -367,10 +369,13 @@ public static class TypeDefGenerator } } + if (isStream) + rt.AppendLine($"[Stream(StreamMode.{f.StreamMode}, Pausable = {f.Pausable.ToString().ToLowerInvariant()})]"); + if (f.IsStatic) { - rt.Append($"[Export] public static AsyncReply<{rtTypeName}> {f.Name}(EpConnection connection"); + rt.Append($"[Export] public static {replyType}<{rtTypeName}> {f.Name}(EpConnection connection"); if (positionalArgs.Length > 0) rt.Append(", " + @@ -383,7 +388,7 @@ public static class TypeDefGenerator } else { - rt.Append($"[Export] public AsyncReply<{rtTypeName}> {f.Name}("); + rt.Append($"[Export] public {replyType}<{rtTypeName}> {f.Name}("); if (positionalArgs.Length > 0) rt.Append( @@ -409,6 +414,15 @@ public static class TypeDefGenerator $"if ({a.Name} != null) args[{a.Index}] = {a.Name};"); } + if (isStream) + { + if (f.IsStatic) + rt.AppendLine($"return connection.StaticStreamCall<{rtTypeName}>({typeDef.Id}UL, {f.Index}, args, StreamMode.{f.StreamMode}); }}"); + else + rt.AppendLine($"return _InvokeStream<{rtTypeName}>({f.Index}, args); }}"); + + continue; + } rt.AppendLine($"var rt = new AsyncReply<{rtTypeName}>();"); diff --git a/Libraries/Esiur/Resource/Attributes/AutoDeliveryAttribute.cs b/Libraries/Esiur/Resource/Attributes/AutoDeliveryAttribute.cs index 26ccbf5..b347f50 100644 --- a/Libraries/Esiur/Resource/Attributes/AutoDeliveryAttribute.cs +++ b/Libraries/Esiur/Resource/Attributes/AutoDeliveryAttribute.cs @@ -38,6 +38,8 @@ namespace Esiur.Resource; AttributeTargets.Field, AllowMultiple = false, Inherited = true)] -public sealed class AutoDeliveredAttribute : Attribute +public class AutoDeliveryAttribute : Attribute { -} \ No newline at end of file + +} + diff --git a/Libraries/Esiur/Resource/Attributes/DefinitionMetadataAttributes.cs b/Libraries/Esiur/Resource/Attributes/DefinitionMetadataAttributes.cs new file mode 100644 index 0000000..e973f5e --- /dev/null +++ b/Libraries/Esiur/Resource/Attributes/DefinitionMetadataAttributes.cs @@ -0,0 +1,163 @@ +using Esiur.Data.Types; +using System; + +namespace Esiur.Resource; + +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct | AttributeTargets.Interface | + AttributeTargets.Enum | AttributeTargets.Method | AttributeTargets.Property | + AttributeTargets.Field | AttributeTargets.Event | AttributeTargets.Parameter, + AllowMultiple = false, Inherited = true)] +public sealed class DescriptionAttribute : Attribute +{ + public string Value { get; } + + public DescriptionAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct | AttributeTargets.Interface | + AttributeTargets.Enum | AttributeTargets.Method | AttributeTargets.Property | + AttributeTargets.Field | AttributeTargets.Event | AttributeTargets.Parameter, + AllowMultiple = true, Inherited = true)] +public sealed class ExampleAttribute : Attribute +{ + public object Value { get; } + + public ExampleAttribute(object value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct | + AttributeTargets.Interface | AttributeTargets.Enum, + AllowMultiple = false, Inherited = true)] +public sealed class CategoryAttribute : Attribute +{ + public string Value { get; } + + public CategoryAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct | + AttributeTargets.Interface | AttributeTargets.Enum, + AllowMultiple = false, Inherited = true)] +public sealed class SinceAttribute : Attribute +{ + public string Value { get; } + + public SinceAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Method | AttributeTargets.Property | + AttributeTargets.Field | AttributeTargets.Event, + AllowMultiple = false, Inherited = true)] +public sealed class TagsAttribute : Attribute +{ + public string[] Values { get; } + + public TagsAttribute(params string[] values) => Values = values; +} + +[AttributeUsage(AttributeTargets.Property | AttributeTargets.Field | + AttributeTargets.Method | AttributeTargets.Event, + AllowMultiple = false, Inherited = true)] +public sealed class UnitAttribute : Attribute +{ + public string Value { get; } + + public UnitAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Property | AttributeTargets.Field | + AttributeTargets.Parameter, + AllowMultiple = false, Inherited = true)] +public sealed class MinimumAttribute : Attribute +{ + public object Value { get; } + + public MinimumAttribute(object value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Property | AttributeTargets.Field | + AttributeTargets.Parameter, + AllowMultiple = false, Inherited = true)] +public sealed class MaximumAttribute : Attribute +{ + public object Value { get; } + + public MaximumAttribute(object value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Property | AttributeTargets.Field | + AttributeTargets.Parameter, + AllowMultiple = true, Inherited = true)] +public sealed class AllowedValueAttribute : Attribute +{ + public object Value { get; } + + public AllowedValueAttribute(object value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Property | AttributeTargets.Field | + AttributeTargets.Parameter, + AllowMultiple = false, Inherited = true)] +public sealed class PatternAttribute : Attribute +{ + public string Value { get; } + + public PatternAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Property | AttributeTargets.Field | + AttributeTargets.Method | AttributeTargets.Event | AttributeTargets.Parameter, + AllowMultiple = false, Inherited = true)] +public sealed class FormatAttribute : Attribute +{ + public string Value { get; } + + public FormatAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Method, + AllowMultiple = true, Inherited = true)] +public sealed class PreconditionAttribute : Attribute +{ + public string Value { get; } + + public PreconditionAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Method, + AllowMultiple = true, Inherited = true)] +public sealed class PostconditionAttribute : Attribute +{ + public string Value { get; } + + public PostconditionAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Method, + AllowMultiple = false, Inherited = true)] +public sealed class EffectsAttribute : Attribute +{ + public OperationEffects Value { get; } + + public EffectsAttribute(OperationEffects value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Method | AttributeTargets.Property | + AttributeTargets.Field | AttributeTargets.Event, + AllowMultiple = true, Inherited = true)] +public sealed class WarningAttribute : Attribute +{ + public string Value { get; } + + public WarningAttribute(string value) => Value = value; +} + +[AttributeUsage(AttributeTargets.Method | AttributeTargets.Property | + AttributeTargets.Field | AttributeTargets.Event, + AllowMultiple = false, Inherited = true)] +public sealed class RelatedMembersAttribute : Attribute +{ + public byte[] Indexes { get; } + + public RelatedMembersAttribute(params byte[] indexes) => Indexes = indexes; +} diff --git a/Libraries/Esiur/Resource/Attributes/HistoricalAttribute.cs b/Libraries/Esiur/Resource/Attributes/HistoricalAttribute.cs index 663d92b..8e2170b 100644 --- a/Libraries/Esiur/Resource/Attributes/HistoricalAttribute.cs +++ b/Libraries/Esiur/Resource/Attributes/HistoricalAttribute.cs @@ -5,12 +5,13 @@ using System.Text; namespace Esiur.Resource { /// - /// Indicates that previous values of an exported property are retained + /// Indicates that previous property values or event occurrences are retained /// and may be fetched remotely. /// [AttributeUsage( AttributeTargets.Property | - AttributeTargets.Field, + AttributeTargets.Field | + AttributeTargets.Event, AllowMultiple = false, Inherited = true)] public sealed class HistoricalAttribute : Attribute diff --git a/Libraries/Esiur/Resource/Attributes/RateControlAttribute.cs b/Libraries/Esiur/Resource/Attributes/RateControlAttribute.cs new file mode 100644 index 0000000..c6fa528 --- /dev/null +++ b/Libraries/Esiur/Resource/Attributes/RateControlAttribute.cs @@ -0,0 +1,26 @@ +using System; + +namespace Esiur.Resource; + +/// +/// Applies a named Warehouse rate policy to an exported function or property setter. +/// +[AttributeUsage( + AttributeTargets.Method | AttributeTargets.Property, + AllowMultiple = false, + Inherited = true)] +public sealed class RateControlAttribute : Attribute +{ + /// + /// Gets the name used to resolve the policy from the owning Warehouse. + /// + public string PolicyName { get; } + + public RateControlAttribute(string policyName) + { + if (string.IsNullOrWhiteSpace(policyName)) + throw new ArgumentException("A rate policy name is required.", nameof(policyName)); + + PolicyName = policyName; + } +} diff --git a/Libraries/Esiur/Resource/Attributes/StreamAttribute.cs b/Libraries/Esiur/Resource/Attributes/StreamAttribute.cs index 29e68dd..f1d5d9f 100644 --- a/Libraries/Esiur/Resource/Attributes/StreamAttribute.cs +++ b/Libraries/Esiur/Resource/Attributes/StreamAttribute.cs @@ -6,7 +6,9 @@ using System.Text; namespace Esiur.Resource { /// - /// Marks an exported function as streaming and specifies its delivery mode. + /// Marks an AsyncReply-returning exported function as streaming and specifies + /// its delivery mode. IEnumerable and IAsyncEnumerable return types infer push + /// and pull mode respectively and do not require this attribute. /// [AttributeUsage( AttributeTargets.Method, diff --git a/Libraries/Esiur/Resource/Attributes/UsageAttribute.cs b/Libraries/Esiur/Resource/Attributes/UsageAttribute.cs index 063c247..2076587 100644 --- a/Libraries/Esiur/Resource/Attributes/UsageAttribute.cs +++ b/Libraries/Esiur/Resource/Attributes/UsageAttribute.cs @@ -2,7 +2,7 @@ using System.Collections.Generic; using System.Text; -namespace Esiur.Resource.Attributes +namespace Esiur.Resource { [AttributeUsage( AttributeTargets.Class | diff --git a/Libraries/Esiur/Resource/Attributes/VolatileAttribute.cs b/Libraries/Esiur/Resource/Attributes/VolatileAttribute.cs index 9386df4..40c383b 100644 --- a/Libraries/Esiur/Resource/Attributes/VolatileAttribute.cs +++ b/Libraries/Esiur/Resource/Attributes/VolatileAttribute.cs @@ -2,7 +2,7 @@ using System.Collections.Generic; using System.Text; -namespace Esiur.Resource.Attributes +namespace Esiur.Resource { /// /// Indicates that an exported property has volatile synchronization diff --git a/Libraries/Esiur/Resource/Warehouse.cs b/Libraries/Esiur/Resource/Warehouse.cs index 44d22d4..67fc885 100644 --- a/Libraries/Esiur/Resource/Warehouse.cs +++ b/Libraries/Esiur/Resource/Warehouse.cs @@ -31,6 +31,7 @@ using Esiur.Protocol; using Esiur.Proxy; using Esiur.Security.Authority; using Esiur.Security.Permissions; +using Esiur.Security.RateLimiting; using Org.BouncyCastle.Asn1.Cms; using System; using System.Collections; @@ -88,6 +89,8 @@ public class Warehouse Map _authenticationProviders = new Map(); List _permissionsManagers = new List(); + readonly ConcurrentDictionary _ratePolicies + = new ConcurrentDictionary(StringComparer.Ordinal); object _typeDefsLock = new object(); @@ -102,6 +105,11 @@ public class Warehouse public KeyList Protocols { get; } = new KeyList(); + /// + /// Runtime settings for this Warehouse. + /// + public WarehouseConfiguration Configuration { get; } + private Regex urlRegex = new Regex(@"^(?:([\S]*)://([^/]*)/?)"); @@ -121,6 +129,41 @@ public class Warehouse _permissionsManagers.Add(manager); } + /// + /// Registers a named rate policy referenced by RateControl attributes. + /// + public void AddRatePolicy(RatePolicy policy) + { + if (policy == null) + throw new ArgumentNullException(nameof(policy)); + if (string.IsNullOrWhiteSpace(policy.Name)) + throw new ArgumentException("The rate policy must have a name.", nameof(policy)); + if (!_ratePolicies.TryAdd(policy.Name, policy)) + throw new InvalidOperationException($"A rate policy named `{policy.Name}` is already registered."); + } + + /// + /// Registers a policy under the supplied name. + /// + public void AddRatePolicy(string name, RatePolicy policy) + { + if (string.IsNullOrWhiteSpace(name)) + throw new ArgumentException("A rate policy name is required.", nameof(name)); + if (policy == null) + throw new ArgumentNullException(nameof(policy)); + + policy.Name = name; + AddRatePolicy(policy); + } + + public RatePolicy? TryGetRatePolicy(string name) + => !string.IsNullOrWhiteSpace(name) && _ratePolicies.TryGetValue(name, out var policy) + ? policy + : null; + + public bool RemoveRatePolicy(string name) + => !string.IsNullOrWhiteSpace(name) && _ratePolicies.TryRemove(name, out _); + public IAuthenticationProvider GetAuthenticationProvider(string name) { if (_authenticationProviders.ContainsKey(name)) @@ -137,8 +180,14 @@ public class Warehouse } - public Warehouse() + public Warehouse() : this(new WarehouseConfiguration()) { + } + + public Warehouse(WarehouseConfiguration configuration) + { + Configuration = configuration ?? throw new ArgumentNullException(nameof(configuration)); + Protocols.Add("EP", async (name, context) => await New(name, context)); diff --git a/Libraries/Esiur/Resource/WarehouseConfiguration.cs b/Libraries/Esiur/Resource/WarehouseConfiguration.cs new file mode 100644 index 0000000..933754c --- /dev/null +++ b/Libraries/Esiur/Resource/WarehouseConfiguration.cs @@ -0,0 +1,33 @@ +using System; + +namespace Esiur.Resource; + +/// +/// Runtime configuration owned by a Warehouse instance. +/// +public sealed class WarehouseConfiguration +{ + public RateControlConfiguration RateControl { get; set; } = new RateControlConfiguration(); +} + +/// +/// Configures how repeated rate-control denials affect an EP connection. +/// +public sealed class RateControlConfiguration +{ + /// + /// Number of denials within that blocks the connection. + /// Set to zero to disable connection blocking. + /// + public int DenialsBeforeConnectionBlock { get; set; } = 5; + + /// + /// Window in which denials are accumulated for connection blocking. + /// + public TimeSpan DenialWindow { get; set; } = TimeSpan.FromMinutes(1); + + /// + /// Delay before a blocked connection is closed, allowing its final error reply to flush. + /// + public TimeSpan ConnectionBlockDelay { get; set; } = TimeSpan.FromMilliseconds(100); +} diff --git a/Libraries/Esiur/Security/RateLimiting/BurstRatePolicy.cs b/Libraries/Esiur/Security/RateLimiting/BurstRatePolicy.cs new file mode 100644 index 0000000..3c319de --- /dev/null +++ b/Libraries/Esiur/Security/RateLimiting/BurstRatePolicy.cs @@ -0,0 +1,140 @@ +using Esiur.Security.Permissions; +using System; +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; + +namespace Esiur.Security.RateLimiting; + +/// +/// Per-connection, per-member token-bucket policy with bounded delayed reservations. +/// +public sealed class BurstRatePolicy : RatePolicy +{ + sealed class Bucket + { + public readonly object Sync = new object(); + public double Tokens; + public long LastTimestamp; + public int Queued; + + public Bucket(double tokens, long timestamp) + { + Tokens = tokens; + LastTimestamp = timestamp; + } + } + + sealed class ConnectionBuckets + { + public readonly ConcurrentDictionary Values + = new ConcurrentDictionary(StringComparer.Ordinal); + } + + readonly ConditionalWeakTable _connections + = new ConditionalWeakTable(); + + /// + /// Number of permits replenished during each period. + /// + public int PermitLimit { get; set; } = 100; + + /// + /// Replenishment period for . + /// + public TimeSpan Period { get; set; } = TimeSpan.FromSeconds(1); + + /// + /// Additional permits available for an immediate burst. + /// + public int BurstLimit { get; set; } + + /// + /// Maximum number of delayed reservations per connection and member. + /// Further requests are denied until queue positions become available. + /// + public int QueueLimit { get; set; } + + public BurstRatePolicy() + { + } + + public BurstRatePolicy(string name) : base(name) + { + } + + public override Ruling Applicable(RateControlContext context) + { + Validate(); + + var capacity = checked(PermitLimit + BurstLimit); + var now = Stopwatch.GetTimestamp(); + var key = $"{(int)context.Action}:{context.Member.Fullname}"; + var buckets = _connections.GetValue(context.Connection, _ => new ConnectionBuckets()); + var bucket = buckets.Values.GetOrAdd(key, _ => new Bucket(capacity, now)); + + TimeSpan delay; + + lock (bucket.Sync) + { + Replenish(bucket, now, capacity); + + if (bucket.Tokens >= 1d) + { + bucket.Tokens -= 1d; + return Ruling.Allowed; + } + + if (QueueLimit == 0 || bucket.Queued >= QueueLimit) + return Ruling.Denied; + + bucket.Tokens -= 1d; + bucket.Queued++; + + var seconds = -bucket.Tokens * Period.TotalSeconds / PermitLimit; + delay = TimeSpan.FromSeconds(Math.Max(0d, seconds)); + context.Delay = delay; + } + + _ = ReleaseQueuePositionAsync(bucket, delay); + return Ruling.Allowed; + } + + void Validate() + { + if (PermitLimit <= 0) + throw new InvalidOperationException("PermitLimit must be greater than zero."); + if (Period <= TimeSpan.Zero) + throw new InvalidOperationException("Period must be greater than zero."); + if (BurstLimit < 0) + throw new InvalidOperationException("BurstLimit cannot be negative."); + if (QueueLimit < 0) + throw new InvalidOperationException("QueueLimit cannot be negative."); + } + + void Replenish(Bucket bucket, long now, int capacity) + { + var elapsedTicks = now - bucket.LastTimestamp; + if (elapsedTicks <= 0) + return; + + var elapsedSeconds = (double)elapsedTicks / Stopwatch.Frequency; + var replenished = elapsedSeconds * PermitLimit / Period.TotalSeconds; + bucket.Tokens = Math.Min(capacity, bucket.Tokens + replenished); + bucket.LastTimestamp = now; + } + + static async Task ReleaseQueuePositionAsync(Bucket bucket, TimeSpan delay) + { + if (delay > TimeSpan.Zero) + await Task.Delay(delay).ConfigureAwait(false); + + lock (bucket.Sync) + { + if (bucket.Queued > 0) + bucket.Queued--; + } + } +} diff --git a/Libraries/Esiur/Security/RateLimiting/RatePolicy.cs b/Libraries/Esiur/Security/RateLimiting/RatePolicy.cs new file mode 100644 index 0000000..15e808f --- /dev/null +++ b/Libraries/Esiur/Security/RateLimiting/RatePolicy.cs @@ -0,0 +1,72 @@ +using Esiur.Data.Types; +using Esiur.Protocol; +using Esiur.Resource; +using Esiur.Security.Authority; +using Esiur.Security.Permissions; +using System; + +namespace Esiur.Security.RateLimiting; + +/// +/// Describes the request currently being evaluated by a rate policy. +/// +public sealed class RateControlContext +{ + public Warehouse Warehouse { get; } + public EpConnection Connection { get; } + public Session Session { get; } + public IResource? Resource { get; } + public MemberDef Member { get; } + public ActionType Action { get; } + + /// + /// Optional delay assigned by a policy to an allowed queued request. + /// + public TimeSpan Delay { get; set; } + + public RateControlContext( + Warehouse warehouse, + EpConnection connection, + Session session, + IResource? resource, + MemberDef member, + ActionType action) + { + Warehouse = warehouse ?? throw new ArgumentNullException(nameof(warehouse)); + Connection = connection ?? throw new ArgumentNullException(nameof(connection)); + Session = session ?? throw new ArgumentNullException(nameof(session)); + Resource = resource; + Member = member ?? throw new ArgumentNullException(nameof(member)); + Action = action; + } +} + +/// +/// Base class for named Warehouse rate-control policies. +/// +public abstract class RatePolicy +{ + /// + /// Name referenced by . + /// + public string Name { get; set; } = string.Empty; + + protected RatePolicy() + { + } + + protected RatePolicy(string name) + { + Name = name; + } + + /// + /// Evaluates a context-free policy. Override this for simple policies. + /// + public virtual Ruling Applicable() => Ruling.DontCare; + + /// + /// Evaluates a request. Context-aware policies should override this overload. + /// + public virtual Ruling Applicable(RateControlContext context) => Applicable(); +} diff --git a/Tests/Features/Functional/MyRecord.cs b/Tests/Features/Functional/MyRecord.cs index 3d2acee..66bca7a 100644 --- a/Tests/Features/Functional/MyRecord.cs +++ b/Tests/Features/Functional/MyRecord.cs @@ -9,7 +9,7 @@ using System.Threading.Tasks; namespace Esiur.Tests.Functional; [Export] - public class MyRecord:IRecord + public class MyRecord : IMyRecord { public string Name { get; set; } public int Id { get; set; } diff --git a/Tests/Features/Functional/MyService.cs b/Tests/Features/Functional/MyService.cs index f3cf53b..6a4cc75 100644 --- a/Tests/Features/Functional/MyService.cs +++ b/Tests/Features/Functional/MyService.cs @@ -1,10 +1,11 @@ using Esiur.Data; using Esiur.Core; +using Esiur.Data.Types; using Esiur.Resource; using System; using System.Collections.Generic; -using System.Text; using System.Threading; +using System.Threading.Tasks; using Esiur.Protocol; #nullable enable @@ -142,7 +143,7 @@ public partial class MyService [Export] public IRecord[] RecordsArray => new IRecord[] { new MyRecord() { Id = 22, Name = "Test", Score = 22.1 } }; [Export] public List RecordsList => new() { new MyRecord() { Id = 22, Name = "Test", Score = 22.1 } }; - [Export] public IMyRecord myrecord { get; set; } + [Export] public IMyRecord myrecord { get; set; } = new MyRecord(); [Export] public MyResource[]? myResources; @@ -214,4 +215,92 @@ public partial class MyService [Export] public static string staticFunction(string name) => $"Hello {name}"; + int _terminatedPullStreams; + + public int TerminatedPullStreams => Volatile.Read(ref _terminatedPullStreams); + + [Export] + [RateControl("standard-call")] + public int RateLimitedCall() => 1; + + [Export] + [RateControl("standard-set")] + public int RateLimitedValue { get; set; } + + [Export] + [Cancellable] + public async IAsyncEnumerable PullRange( + int start, + int count, + int delayMilliseconds, + InvocationContext context) + { + for (var i = 0; i < count; i++) + { + await Task.Delay(Math.Max(0, delayMilliseconds), context.CancellationToken); + yield return start + i; + } + } + + [Export] + [Cancellable] + public async IAsyncEnumerable PullForever( + int delayMilliseconds, + InvocationContext context) + { + var value = 0; + + try + { + while (true) + { + await Task.Delay(Math.Max(1, delayMilliseconds), context.CancellationToken); + yield return value++; + } + } + finally + { + Interlocked.Increment(ref _terminatedPullStreams); + } + } + + [Export] + [Cancellable] + [Stream(StreamMode.Push, Pausable = true)] + public AsyncReply PushSequence( + int count, + int delayMilliseconds, + InvocationContext context) + { + var reply = new AsyncReply(); + + _ = Task.Run(async () => + { + try + { + await Task.Delay(50, context.CancellationToken); + + for (var i = 0; i < count; i++) + { + await context.WaitWhileHaltedAsync(); + context.CancellationToken.ThrowIfCancellationRequested(); + reply.TriggerChunk(i); + await Task.Delay(Math.Max(1, delayMilliseconds), context.CancellationToken); + } + + reply.Trigger(0); + } + catch (OperationCanceledException) + { + // TerminateExecution completes the remote stream. + } + catch (Exception exception) + { + reply.TriggerError(exception); + } + }); + + return reply; + } + } diff --git a/Tests/Features/Functional/Program.cs b/Tests/Features/Functional/Program.cs index e7a6c24..3cf1a43 100644 --- a/Tests/Features/Functional/Program.cs +++ b/Tests/Features/Functional/Program.cs @@ -1,4 +1,4 @@ -/* +/* Copyright (c) 2017-2026 Ahmed Kh. Zamil @@ -22,337 +22,338 @@ SOFTWARE. */ -using Esiur.Data; using Esiur.Core; -using Esiur.Net.Http; -using Esiur.Net.Sockets; +using Esiur.Data; +using Esiur.Data.Types; +using Esiur.Protocol; using Esiur.Resource; +using Esiur.Security.Authority; using Esiur.Security.Permissions; +using Esiur.Security.RateLimiting; using Esiur.Stores; using System; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; -using Esiur.Security.Integrity; -using System.Linq; -using Esiur.Data.Types; -using System.Collections; -using System.Runtime.CompilerServices; -using Esiur.Proxy; -using System.Text.RegularExpressions; -using System.Collections.Generic; -using System.Reflection; -using System.Security.Cryptography; -using System.Text; -using Esiur.Security.Cryptography; -using Esiur.Security.Membership; -using Esiur.Net.Packets; -using System.Numerics; -using Esiur.Protocol; -using Esiur.Security.Authority.Providers; -using Esiur.Security.Authority; +#nullable enable namespace Esiur.Tests.Functional; - -class Program +internal static class Program { - - static void TestSerialization(object x, EpConnection connection = null) + static async Task Main() { + var serverWarehouse = new Warehouse(); + var clientWarehouse = new Warehouse(); + EpConnection? connection = null; - var d = Codec.Compose(x, Warehouse.Default, connection); - // var rr = DC.ToHex(y); - - var y = Codec.ParseSync(d, 0, Warehouse.Default); - Console.WriteLine($"{x.GetType().Name}: {x} == {y}, {d.ToHex()}"); - } - - - [Export] - public class StudentRecord : IRecord - { - public string Name { get; set; } - public byte Grade { get; set; } - } - public enum LogLevel : int - { - Debug, - Warning, - Error, - } - - static async Task Main(string[] args) - { - //TestSerialization("Hello"); - //TestSerialization(10); - //TestSerialization(10.1); - //TestSerialization(10.1d); - //TestSerialization((byte)1); - //TestSerialization((byte)2); - //TestSerialization(new int[] { 1, 2, 3, 4 }); - //var x = LogLevel.Warning; - - //TestSerialization(LogLevel.Warning); - - //TestSerialization(new Map - //{ - // ["C++"] = 1, - // ["C#"] = 2, - // ["JS"] = null - //}); - - - - //TestSerialization(new StudentRecord() { Name = "Ali", Grade = 90 }); - - //var tn = Encoding.UTF8.GetBytes("Test.StudentRecord"); - //var hash = System.Security.Cryptography.SHA256.Create().ComputeHash(tn).Clip(0, 16); - //hash[6] = (byte)((hash[6] & 0xF) | 0x80); - //hash[8] = (byte)((hash[8] & 0xF) | 0x80); - - //var g = new UUID(hash); - - //Console.WriteLine(g); - - - - //var a = new ECDH(); - //var b = new ECDH(); - - //var apk = a.GetPublicKey(); - //var bpk = b.GetPublicKey(); - - //var ska = a.ComputeSharedKey(bpk); - //var skb = b.ComputeSharedKey(apk); - - //Console.WriteLine(ska.ToHex()); - //Console.WriteLine(skb.ToHex()); - - //// Simple membership provider - //var membership = new SimpleMembership() { GuestsAllowed = true }; - - //membership.AddUser("user", "123456", new SimpleMembership.QuestionAnswer[0]); - //membership.AddUser("admin", "admin", new SimpleMembership.QuestionAnswer[] - //{ - // new SimpleMembership.QuestionAnswer() - // { - // Question = "What is 5+5", - // Answer = 10, - // Hashed = true, - // } - //}); - - var wh = new Warehouse(); - wh.RegisterAuthenticationProvider(new ServerAuthenticationProvider()); - - // Create stores to keep objects. - var system = await wh.Put("sys", new MemoryStore()); - var server = await wh.Put("sys/server", new EpServer() + try { - AllowedAuthenticationProviders = new string[] { "hash" }, + var port = FindAvailablePort(); + var service = await StartServer(serverWarehouse, port); + + connection = await ConnectClient(clientWarehouse, port); + var remote = await connection.Get("sys/service") as EpResource + ?? throw new InvalidOperationException("Remote service was not found."); + + await RunCoreScenarios(connection, service, remote); + await RunRateControlScenarios(remote); + await RunStreamingScenarios(service, remote); + + Console.WriteLine(); + Console.WriteLine("All functional scenarios passed."); + } + finally + { + try { connection?.Destroy(); } catch { } + try { await clientWarehouse.Close(); } catch { } + try { await serverWarehouse.Close(); } catch { } + } + } + + static async Task StartServer(Warehouse warehouse, ushort port) + { + warehouse.RegisterAuthenticationProvider(new ServerAuthenticationProvider()); + warehouse.Configuration.RateControl.DenialsBeforeConnectionBlock = 10; + warehouse.AddRatePolicy(new BurstRatePolicy("standard-call") + { + PermitLimit = 1, + Period = TimeSpan.FromMinutes(1), + QueueLimit = 0, + }); + warehouse.AddRatePolicy(new BurstRatePolicy("standard-set") + { + PermitLimit = 1, + Period = TimeSpan.FromMinutes(1), + QueueLimit = 0, }); - - var web = await wh.Put("sys/web", new HttpServer() { Port = 8088 }); - - var service = await wh.Put("sys/service", new MyService()); - var res1 = await wh.Put("sys/service/r1", new MyResource() { Description = "Testing 1", CategoryId = 10 }); - var res2 = await wh.Put("sys/service/r2", new MyResource() { Description = "Testing 2", CategoryId = 11 }); - var res3 = await wh.Put("sys/service/c1", new MyChildResource() { ChildName = "Child 1", Description = "Child Testing 3", CategoryId = 12 }); - var res4 = await wh.Put("sys/service/c2", new MyChildResource() { ChildName = "Child 2 Destroy", Description = "Testing Destroy Handler", CategoryId = 12 }); - - //TestSerialization(res1); - - server.MapCall("Hello", (string msg, DateTime time, EpConnection sender) => + await warehouse.Put("sys", new MemoryStore()); + var server = await warehouse.Put("sys/server", new EpServer { - Console.WriteLine(msg); - return "Hi " + DateTime.UtcNow; - }).MapCall("temp", () => res4); - - service.Resource = res1; - service.ChildResource = res3; - service.Resources = new MyResource[] { res1, res2, res1, res3 }; - service.MyResources = new MyResource[] { res1, res2, res3, res4 }; - - //web.MapGet("/{action}/{age}", (int age, string action, HTTPConnection sender) => - //{ - // Console.WriteLine($"AGE: {age} ACTION: {action}"); - - // sender.Response.Number = Esiur.Net.Packets.HTTPResponsePacket.ResponseCode.NotFound; - // sender.Send("Not found"); - //}); - - web.MapGet("/", (HttpConnection sender) => - { - sender.Send("Hello"); + Port = port, + AllowedAuthenticationProviders = new[] { "hash" }, }); - await wh.Open(); + var service = await warehouse.Put("sys/service", new MyService()); + service.Instance!.Managers.Add(new AllowPropertySetPermissions()); + var resource1 = await warehouse.Put("sys/service/r1", new MyResource + { + Description = "Testing 1", + CategoryId = 10, + }); + var resource2 = await warehouse.Put("sys/service/r2", new MyResource + { + Description = "Testing 2", + CategoryId = 11, + }); + var child1 = await warehouse.Put("sys/service/c1", new MyChildResource + { + ChildName = "Child 1", + Description = "Child Testing 3", + CategoryId = 12, + }); + var child2 = await warehouse.Put("sys/service/c2", new MyChildResource + { + ChildName = "Child 2", + Description = "Testing lifecycle controls", + CategoryId = 12, + }); + service.Resource = resource1; + service.ChildResource = child1; + service.Resources = new MyResource[] { resource1, resource2, resource1, child1 }; + service.MyResources = new MyResource[] { resource1, resource2, child1, child2 }; - //var sc = service.GetGenericRecord(); - //var d = Codec.Compose(sc, Warehouse.Default, null); + server.MapCall("Hello", (string message, DateTime _, EpConnection __) => $"Hi {message}"); + server.MapCall("temp", () => child2); - // Start testing - TestClient(service); + await warehouse.Open(); + return service; } - - // AuthorizationRequest, AsyncReply - static AsyncReply Authenticator(AuthorizationRequest x) + static async Task ConnectClient(Warehouse warehouse, ushort port) { - Console.WriteLine($"Authenticator: {x.Clue}"); + warehouse.RegisterAuthenticationProvider(new ClientAuthenticationProvider()); - var format = x.RequiredFormat; - - if (format == EpAuthPacketIAuthFormat.Number) - return new AsyncReply(Convert.ToInt32(10)); - else if (format == EpAuthPacketIAuthFormat.Text) - return new AsyncReply(Console.ReadLine().Trim()); - - throw new NotImplementedException("Not supported format."); - } - - private static async void TestClient(IResource local) - { - var wh = new Warehouse(); - var auth = new ClientAuthenticationProvider(); - wh.RegisterAuthenticationProvider(auth); - - var con = await wh.Get("ep://localhost", new EpConnectionContext + return await warehouse.Get($"ep://localhost:{port}", new EpConnectionContext { AuthenticationMode = AuthenticationMode.InitializerIdentity, - AutoReconnect = true, + AutoReconnect = false, Identity = "tester", AuthenticationProtocol = "hash", Domain = "test", }); - - - dynamic remote = await con.Get("sys/service"); - - TestObjectProps(local, remote); - - //return; - - //return; - - Console.WriteLine("OK"); - - perodicTimer = new Timer(new TimerCallback(perodicTimerElapsed), remote, 0, 1000); - - var pcall = await con.Call("Hello", "whats up ?", DateTime.UtcNow); - - var temp = await con.Call("temp"); - Console.WriteLine("Temp: " + temp.GetHashCode()); - - - - - var opt = await remote.Optional(new { a1 = 22, a2 = 33, a4 = "What?" }); - Console.WriteLine(opt); - - var hello = await remote.AsyncHello(); - - await remote.Void(); - await remote.Connection("ss", 33); - await remote.ConnectionOptional("Test 2", 88); - var rt = await remote.Optional("Optiona", 311); - Console.WriteLine(rt); - - var t2 = await remote.GetTuple2(1, "A"); - Console.WriteLine(t2); - var t3 = await remote.GetTuple3(1, "A", 1.3); - Console.WriteLine(t3); - var t4 = await remote.GetTuple4(1, "A", 1.3, true); - Console.WriteLine(t4); - - remote.StringEvent += new EpResourceEvent((sender, args) => - Console.WriteLine($"StringEvent {args}") - ); - - remote.ArrayEvent += new EpResourceEvent((sender, args) => - Console.WriteLine($"ArrayEvent {args}") - ); - - await remote.InvokeEvents("Hello"); - - - - //var path = TemplateGenerator.GetTemplate("EP://localhost/sys/service", "Generated"); - - //Console.WriteLine(path); - - } - static async void perodicTimerElapsed(object state) + static async Task RunCoreScenarios(EpConnection connection, MyService local, EpResource remote) + { + Console.WriteLine("Core RPC and serialization"); + ReportProperties(local, remote); + + dynamic api = remote; + + var procedureResult = (string)await connection.Call("Hello", "functional", DateTime.UtcNow); + Require(procedureResult == "Hi functional", "Procedure call returned an unexpected value."); + + var temporaryResource = (IResource)await connection.Call("temp"); + Require( + temporaryResource is EpResource && + temporaryResource.Instance?.Definition.Name.EndsWith(nameof(MyChildResource), StringComparison.Ordinal) == true, + "Procedure call did not return the expected remote resource type."); + + var optional = await api.Optional(new { a1 = 22, a2 = 33, a4 = "What?" }); + Require(optional is double, "Optional argument invocation failed."); + + var hello = await api.AsyncHello(); + Require(hello is not null, "AsyncReply invocation returned null."); + + await api.Void(); + await api.Connection("connection", 33); + await api.ConnectionOptional("optional connection", 88); + + var tuple = await api.GetTuple4(1, "A", 1.3, true); + Require(tuple is not null, "Tuple invocation returned null."); + + var eventReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + EpResourceEvent handler = (_, value) => eventReceived.TrySetResult((string)value); + await remote.Subscribe(nameof(MyService.StringEvent)); + api.StringEvent += handler; + + await api.InvokeEvents("event payload"); + var eventValue = await eventReceived.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Require(eventValue == "event payload", "Remote event payload did not round-trip."); + + api.StringEvent -= handler; + await remote.Unsubscribe(nameof(MyService.StringEvent)); + Console.WriteLine(" PASS core RPC, optional arguments, tuples, and events"); + } + + static async Task RunStreamingScenarios(MyService local, EpResource remote) + { + Console.WriteLine("Streaming and execution controls"); + + var pull = InvokeStream(remote, nameof(MyService.PullRange), 10, 5, 5); + var pulledValues = new List(); + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5))) + { + await foreach (var value in pull.WithCancellation(timeout.Token)) + pulledValues.Add(value); + } + + Require( + pulledValues.SequenceEqual(new[] { 10, 11, 12, 13, 14 }), + "PullStream did not preserve item order or backpressure."); + Console.WriteLine(" PASS PullStream / IAsyncEnumerable"); + + var terminatedBefore = local.TerminatedPullStreams; + var infinite = InvokeStream(remote, nameof(MyService.PullForever), 5); + var infiniteEnumerator = infinite.GetAsyncEnumerator(); + + Require(await infiniteEnumerator.MoveNextAsync(), "Infinite pull stream returned no first item."); + Require(await infiniteEnumerator.MoveNextAsync(), "Infinite pull stream returned no second item."); + await infiniteEnumerator.DisposeAsync(); + + await WaitUntil( + () => local.TerminatedPullStreams == terminatedBefore + 1, + TimeSpan.FromSeconds(5)); + Require(infinite.Completed, "TerminateExecution did not complete the client stream."); + Console.WriteLine(" PASS TerminateExecution / enumerator disposal"); + + var push = InvokeStream(remote, nameof(MyService.PushSequence), 4, 150); + var pushEnumerator = push.GetAsyncEnumerator(); + var pushedValues = new List(); + + Require(await pushEnumerator.MoveNextAsync(), "Push stream returned no first item."); + pushedValues.Add(pushEnumerator.Current); + + await push.Halt(); + var haltedMove = pushEnumerator.MoveNextAsync().AsTask(); + await Task.Delay(250); + Require(!haltedMove.IsCompleted, "HaltExecution did not pause the producer."); + + await push.Resume(); + Require( + await haltedMove.WaitAsync(TimeSpan.FromSeconds(5)), + "ResumeExecution did not resume the producer."); + pushedValues.Add(pushEnumerator.Current); + + while (await pushEnumerator.MoveNextAsync()) + pushedValues.Add(pushEnumerator.Current); + + Require( + pushedValues.SequenceEqual(new[] { 0, 1, 2, 3 }), + "Push stream lost or reordered items across halt/resume."); + Console.WriteLine(" PASS HaltExecution / ResumeExecution"); + } + + static async Task RunRateControlScenarios(EpResource remote) + { + Console.WriteLine("Rate control"); + + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(MyService.RateLimitedCall)) + ?? throw new InvalidOperationException("Rate-limited function was not found."); + var property = remote.Instance.Definition.GetPropertyDefByName(nameof(MyService.RateLimitedValue)) + ?? throw new InvalidOperationException("Rate-limited property was not found."); + + await remote._Invoke(function.Index, Array.Empty()); + await ExpectRateLimit(() => remote._Invoke(function.Index, Array.Empty())); + + await remote.SetResourcePropertyAsync(property.Index, 1); + await ExpectRateLimit(() => remote.SetResourcePropertyAsync(property.Index, 2)); + + Console.WriteLine(" PASS function-call and property-set policies"); + } + + static async Task ExpectRateLimit(Func action) { - GC.Collect(); try { - dynamic remote = state; - await remote.InvokeEvents("Hello"); - - Console.WriteLine("Perodic : " + await remote.AsyncHello()); + await action(); } - catch (Exception ex) + catch (AsyncException exception) when (exception.Code == ExceptionCode.RateLimitExceeded) { - Console.WriteLine("Perodic : " + ex.ToString()); + return; } + + throw new InvalidOperationException("The request was expected to be rate limited."); } - static Timer perodicTimer; - - static void TestObjectProps(IResource local, EpResource remote) + static AsyncStreamReply InvokeStream( + EpResource resource, + string functionName, + params object[] arguments) { + var function = resource.Instance.Definition.GetFunctionDefByName(functionName) + ?? throw new InvalidOperationException($"Function `{functionName}` was not found."); - foreach (var pt in local.Instance.Definition.Properties) - { - - var lv = pt.PropertyInfo.GetValue(local); - object v; - var rv = remote.TryGetPropertyValue(pt.Index, out v); - if (!rv) - Console.WriteLine($" ** {pt.Name} Failed"); - else - Console.WriteLine($"{pt.Name} {GetString(lv)} == {GetString(v)}"); - } + var indexedArguments = new Map(); + for (byte i = 0; i < arguments.Length; i++) + indexedArguments[i] = arguments[i]; + return resource._InvokeStream(function.Index, indexedArguments); } - static string GetString(object value) + static void ReportProperties(MyService local, EpResource remote) { - if (value == null) - return "NULL"; + var compared = 0; + var definition = local.Instance?.Definition + ?? throw new InvalidOperationException("Local service was not initialized."); - var t = value.GetType(); - var nt = Nullable.GetUnderlyingType(t); - if (nt != null) - t = nt; - if (t.IsArray) + foreach (var property in definition.Properties) { - var ar = (Array)value; - if (ar.Length == 0) - return "[]"; - var rt = "["; - for (var i = 0; i < ar.Length - 1; i++) - rt += GetString(ar.GetValue(i)) + ","; - rt += GetString(ar.GetValue(ar.Length - 1)) + "]"; + if (!remote.TryGetPropertyValue(property.Index, out _)) + throw new InvalidOperationException($"Remote property `{property.Name}` was not attached."); - return rt; - } - else if (value is Record) - { - return value.ToString(); - } - else if (value is IRecord) - { - return "{" + String.Join(", ", t.GetProperties().Select(x => x.Name + ": " + x.GetValue(value))) + "}"; + compared++; } - else - return value.ToString(); + Console.WriteLine($" Attached {compared} exported properties"); } -} \ No newline at end of file + + static async Task WaitUntil(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition()) + { + if (DateTime.UtcNow >= deadline) + throw new TimeoutException("Timed out waiting for the functional condition."); + + await Task.Delay(10); + } + } + + static void Require(bool condition, string message) + { + if (!condition) + throw new InvalidOperationException(message); + } + + static ushort FindAvailablePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = (ushort)((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + sealed class AllowPropertySetPermissions : IPermissionsManager + { + public Map Settings { get; } = new(); + + public Ruling Applicable( + IResource resource, + Session session, + ActionType action, + MemberDef member, + object inquirer = null!) + => action == ActionType.SetProperty ? Ruling.Allowed : Ruling.DontCare; + + public bool Initialize(Map settings, IResource resource) => true; + } +} diff --git a/Tests/Features/Functional/README.md b/Tests/Features/Functional/README.md new file mode 100644 index 0000000..a3e48cf --- /dev/null +++ b/Tests/Features/Functional/README.md @@ -0,0 +1,43 @@ +# Esiur functional smoke test + +This project starts an in-process Esiur server and an authenticated client, +runs its scenarios over a real loopback connection, and exits with a non-zero +status if any assertion fails. + +Coverage includes: + +- resource attachment and property serialization; +- procedure and resource calls, optional arguments, tuples, and events; +- named rate policies for function calls and property setters; +- rate-limit denial propagation to callers; +- pull streams backed by `IAsyncEnumerable`; +- `TerminateExecution` through async-enumerator disposal; +- `HaltExecution` and `ResumeExecution` for a cooperative push stream. + +Run it from the repository root: + +```powershell +dotnet run --project Tests\Features\Functional\Esiur.Tests.Functional.csproj +``` + +The runner selects an available loopback port and shuts down its client, +server, and warehouses when complete. + +Rate policies are registered by name on the server Warehouse: + +```csharp +warehouse.AddRatePolicy(new BurstRatePolicy("standard-call") +{ + PermitLimit = 100, + Period = TimeSpan.FromSeconds(1), + BurstLimit = 20, + QueueLimit = 50, +}); + +warehouse.Configuration.RateControl.DenialsBeforeConnectionBlock = 5; + +[RateControl("standard-call")] +public void Call() +{ +} +``` diff --git a/Tests/Unit/AsyncStreamReplyTests.cs b/Tests/Unit/AsyncStreamReplyTests.cs new file mode 100644 index 0000000..84d22d9 --- /dev/null +++ b/Tests/Unit/AsyncStreamReplyTests.cs @@ -0,0 +1,159 @@ +using Esiur.Core; +using Esiur.Data.Types; +using System.Runtime.CompilerServices; + +namespace Esiur.Tests.Unit; + +public class AsyncStreamReplyTests +{ + [Fact] + public async Task PullStream_RequestsOneItemPerMoveNext() + { + var pulls = 0; + var stream = CreateStream(StreamMode.Pull, pull: () => + { + pulls++; + return CompletedReply(); + }); + + stream.TriggerStreamStarted(); + var enumerator = stream.GetAsyncEnumerator(); + + var firstMove = enumerator.MoveNextAsync().AsTask(); + Assert.Equal(1, pulls); + Assert.False(firstMove.IsCompleted); + + stream.TriggerChunk(17); + Assert.True(await firstMove); + Assert.Equal(17, enumerator.Current); + + var secondMove = enumerator.MoveNextAsync().AsTask(); + Assert.Equal(2, pulls); + stream.TriggerStreamCompleted(); + + Assert.False(await secondMove); + await enumerator.DisposeAsync(); + } + + [Fact] + public async Task DisposingStream_TerminatesRemoteExecutionOnce() + { + var terminations = 0; + var stream = CreateStream(StreamMode.Pull, terminate: () => + { + terminations++; + return CompletedReply(); + }); + + stream.TriggerStreamStarted(); + var enumerator = stream.GetAsyncEnumerator(); + + await enumerator.DisposeAsync(); + await enumerator.DisposeAsync(); + + Assert.Equal(1, terminations); + Assert.True(stream.Completed); + } + + [Fact] + public void LifecycleControls_AreSentThroughStreamReply() + { + var halts = 0; + var resumes = 0; + var stream = CreateStream( + StreamMode.Push, + halt: () => + { + halts++; + return CompletedReply(); + }, + resume: () => + { + resumes++; + return CompletedReply(); + }); + + stream.TriggerStreamStarted(); + stream.Halt(); + stream.Resume(); + + Assert.Equal(1, halts); + Assert.Equal(1, resumes); + Assert.Throws(() => stream.Pull()); + } + + [Fact] + public async Task InvocationContext_PullsGenericAsyncEnumerableAndCancelsIt() + { + var disposed = false; + var context = new InvocationContext(null!, 1); + context.InitializeStream(StreamMode.Pull, pausable: false); + + Assert.True(context.SetAsyncEnumerable(Values(() => disposed = true))); + + var first = await context.PullAsync(); + var second = await context.PullAsync(); + + Assert.True(first.HasValue); + Assert.Equal(1, first.Value); + Assert.True(second.HasValue); + Assert.Equal(2, second.Value); + + await context.TerminateAsync(); + + Assert.True(context.CancellationToken.IsCancellationRequested); + Assert.True(disposed); + } + + [Fact] + public async Task InvocationContext_HaltAndResumeGatePushEnumeration() + { + var context = new InvocationContext(null!, 1); + context.InitializeStream(StreamMode.Push, pausable: true); + context.SetEnumerable(new[] { 3, 4 }); + + Assert.True(context.Halt()); + var move = context.MoveNextAsync(); + Assert.False(move.IsCompleted); + + Assert.True(context.Resume()); + var item = await move; + + Assert.True(item.HasValue); + Assert.Equal(3, item.Value); + await context.EndAsync(); + } + + static AsyncStreamReply CreateStream( + StreamMode mode, + Func? pull = null, + Func? terminate = null, + Func? halt = null, + Func? resume = null) + => new( + mode, + pull ?? CompletedReply, + terminate ?? CompletedReply, + halt ?? CompletedReply, + resume ?? CompletedReply); + + static AsyncReply CompletedReply() => new((object)null!); + + static async IAsyncEnumerable Values( + Action onDisposed, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + try + { + yield return 1; + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + yield return 2; + await Task.Delay(Timeout.Infinite, cancellationToken); + } + finally + { + onDisposed(); + } + } +} diff --git a/Tests/Unit/Integration/AsyncStreamIntegrationTests.cs b/Tests/Unit/Integration/AsyncStreamIntegrationTests.cs new file mode 100644 index 0000000..880ae21 --- /dev/null +++ b/Tests/Unit/Integration/AsyncStreamIntegrationTests.cs @@ -0,0 +1,80 @@ +using Esiur.Core; +using Esiur.Data; +using Esiur.Protocol; + +namespace Esiur.Tests.Unit.Integration; + +[Collection("Integration")] +public class AsyncStreamIntegrationTests +{ + [Fact] + public async Task AsyncEnumerable_IsPulledAcrossProtocol() + { + await using var cluster = await StartCluster().WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await Task.Run(async () => + (EpResource)await cluster.Connection.Get("sys/stream")) + .WaitAsync(TimeSpan.FromSeconds(10)); + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(StreamResource.Numbers)); + var stream = remote._InvokeStream( + function.Index, + new Map { [0] = 4 }); + + var values = new List(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + await foreach (var value in stream.WithCancellation(timeout.Token)) + values.Add(value); + + Assert.Equal(new[] { 0, 1, 2, 3 }, values); + Assert.True(stream.Completed); + } + + [Fact] + public async Task DisposingAsyncEnumerable_TerminatesRemoteExecution() + { + await using var cluster = await StartCluster().WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await Task.Run(async () => + (EpResource)await cluster.Connection.Get("sys/stream")) + .WaitAsync(TimeSpan.FromSeconds(10)); + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(StreamResource.Infinite)); + var stream = remote._InvokeStream(function.Index, Array.Empty()); + var enumerator = stream.GetAsyncEnumerator(); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(0, enumerator.Current); + + await enumerator.DisposeAsync(); + Assert.True(stream.Completed); + } + + [Fact] + public async Task PausablePushStream_HaltsAndResumesAcrossProtocol() + { + await using var cluster = await StartCluster().WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await Task.Run(async () => + (EpResource)await cluster.Connection.Get("sys/stream")) + .WaitAsync(TimeSpan.FromSeconds(10)); + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(StreamResource.Pausable)); + var stream = remote._InvokeStream(function.Index, Array.Empty()); + var enumerator = stream.GetAsyncEnumerator(); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(0, enumerator.Current); + + await stream.Halt(); + var haltedMove = enumerator.MoveNextAsync().AsTask(); + await Task.Delay(200); + Assert.False(haltedMove.IsCompleted); + + await stream.Resume(); + Assert.True(await haltedMove.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Equal(1, enumerator.Current); + + await enumerator.DisposeAsync(); + } + + static Task StartCluster() + => IntegrationCluster.StartAsync(async warehouse => + { + await warehouse.Put("sys/stream", new StreamResource()); + }); +} diff --git a/Tests/Unit/Integration/RateControlIntegrationTests.cs b/Tests/Unit/Integration/RateControlIntegrationTests.cs new file mode 100644 index 0000000..a89030e --- /dev/null +++ b/Tests/Unit/Integration/RateControlIntegrationTests.cs @@ -0,0 +1,141 @@ +using Esiur.Core; +using Esiur.Data; +using Esiur.Data.Types; +using Esiur.Protocol; +using Esiur.Resource; +using Esiur.Security.Authority; +using Esiur.Security.Permissions; +using Esiur.Security.RateLimiting; +using System.Diagnostics; + +namespace Esiur.Tests.Unit.Integration; + +[Collection("Integration")] +public class RateControlIntegrationTests +{ + [Fact] + public async Task FunctionCalls_AreDeniedWhenPolicyIsExhausted() + { + await using var cluster = await StartCluster().WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(RateLimitedResource.Call)); + + var first = await remote._Invoke(function.Index, Array.Empty()); + var exception = await Assert.ThrowsAsync(async () => + await remote._Invoke(function.Index, Array.Empty())); + + Assert.Equal(1, Convert.ToInt32(first)); + Assert.Equal(ExceptionCode.RateLimitExceeded, exception.Code); + } + + [Fact] + public async Task PropertySets_AreDeniedWhenPolicyIsExhausted() + { + await using var cluster = await StartCluster().WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var property = remote.Instance.Definition.GetPropertyDefByName(nameof(RateLimitedResource.Value)); + + await remote.SetResourcePropertyAsync(property.Index, 10); + var exception = await Assert.ThrowsAsync(async () => + await remote.SetResourcePropertyAsync(property.Index, 20)); + + Assert.Equal(ExceptionCode.RateLimitExceeded, exception.Code); + } + + [Fact] + public async Task RepeatedDenials_BlockTheConnection() + { + await using var cluster = await StartCluster(denialsBeforeBlock: 1) + .WaitAsync(TimeSpan.FromSeconds(10)); + cluster.Connection.AutoReconnect = false; + var remote = await GetRemote(cluster); + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(RateLimitedResource.Call)); + + await remote._Invoke(function.Index, Array.Empty()); + var exception = await Assert.ThrowsAsync(async () => + await remote._Invoke(function.Index, Array.Empty())); + + Assert.Equal(ExceptionCode.RateLimitExceeded, exception.Code); + await WaitUntilAsync(() => !cluster.Connection.IsConnected, TimeSpan.FromSeconds(3)); + Assert.False(cluster.Connection.IsConnected); + } + + [Fact] + public async Task QueuedCalls_AreDelayedAndQueueOverflowIsDenied() + { + await using var cluster = await StartCluster( + queueLimit: 1, + period: TimeSpan.FromMilliseconds(250)).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var function = remote.Instance.Definition.GetFunctionDefByName(nameof(RateLimitedResource.Call)); + + await remote._Invoke(function.Index, Array.Empty()); + + var stopwatch = Stopwatch.StartNew(); + var queued = remote._Invoke(function.Index, Array.Empty()); + var exception = await Assert.ThrowsAsync(async () => + await remote._Invoke(function.Index, Array.Empty())); + var result = await queued; + + Assert.Equal(ExceptionCode.RateLimitExceeded, exception.Code); + Assert.Equal(2, Convert.ToInt32(result)); + Assert.True(stopwatch.Elapsed >= TimeSpan.FromMilliseconds(150)); + } + + static Task StartCluster( + int denialsBeforeBlock = 10, + int queueLimit = 0, + TimeSpan? period = null) + => IntegrationCluster.StartAsync(async warehouse => + { + warehouse.Configuration.RateControl.DenialsBeforeConnectionBlock = denialsBeforeBlock; + warehouse.Configuration.RateControl.ConnectionBlockDelay = TimeSpan.FromMilliseconds(100); + + warehouse.AddRatePolicy(new BurstRatePolicy("standard-call") + { + PermitLimit = 1, + Period = period ?? TimeSpan.FromMinutes(1), + QueueLimit = queueLimit, + }); + warehouse.AddRatePolicy(new BurstRatePolicy("standard-set") + { + PermitLimit = 1, + Period = period ?? TimeSpan.FromMinutes(1), + QueueLimit = queueLimit, + }); + + var resource = await warehouse.Put("sys/rate", new RateLimitedResource()); + resource.Instance!.Managers.Add(new AllowPropertySetPermissions()); + }); + + static async Task GetRemote(IntegrationCluster cluster) + => await Task.Run(async () => + (EpResource)await cluster.Connection.Get("sys/rate")) + .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); + } + } + + sealed class AllowPropertySetPermissions : IPermissionsManager + { + public Map Settings { get; } = new(); + + public Ruling Applicable( + IResource resource, + Session session, + ActionType action, + MemberDef member, + object inquirer = null!) + => action == ActionType.SetProperty ? Ruling.Allowed : Ruling.DontCare; + + public bool Initialize(Map settings, IResource resource) => true; + } +} diff --git a/Tests/Unit/Integration/RateLimitedResource.cs b/Tests/Unit/Integration/RateLimitedResource.cs new file mode 100644 index 0000000..38e7964 --- /dev/null +++ b/Tests/Unit/Integration/RateLimitedResource.cs @@ -0,0 +1,23 @@ +using Esiur.Resource; +using System.Threading; + +namespace Esiur.Tests.Unit.Integration; + +[Resource] +public partial class RateLimitedResource +{ + int _callCount; + int _value; + + [Export] + [RateControl("standard-call")] + public int Call() => Interlocked.Increment(ref _callCount); + + [Export] + [RateControl("standard-set")] + public int Value + { + get => _value; + set => _value = value; + } +} diff --git a/Tests/Unit/Integration/StreamResource.cs b/Tests/Unit/Integration/StreamResource.cs new file mode 100644 index 0000000..b7ab681 --- /dev/null +++ b/Tests/Unit/Integration/StreamResource.cs @@ -0,0 +1,53 @@ +using Esiur.Core; +using Esiur.Data.Types; +using Esiur.Resource; + +namespace Esiur.Tests.Unit.Integration; + +[Resource] +public partial class StreamResource +{ + [Export] + public async IAsyncEnumerable Numbers(int count, InvocationContext context) + { + for (var i = 0; i < count; i++) + { + await Task.Delay(5, context.CancellationToken); + yield return i; + } + } + + [Export] + public async IAsyncEnumerable Infinite(InvocationContext context) + { + var value = 0; + while (true) + { + await Task.Delay(5, context.CancellationToken); + yield return value++; + } + } + + [Export] + [Stream(StreamMode.Push, Pausable = true)] + public AsyncReply Pausable(InvocationContext context) + { + var reply = new AsyncReply(); + + Task.Run(async () => + { + await Task.Delay(100); + + for (var i = 0; i < 3; i++) + { + await context.WaitWhileHaltedAsync(); + reply.TriggerChunk(i); + await Task.Delay(100); + } + + reply.Trigger(0); + }); + + return reply; + } +} diff --git a/Tests/Unit/RatePolicyTests.cs b/Tests/Unit/RatePolicyTests.cs new file mode 100644 index 0000000..3485219 --- /dev/null +++ b/Tests/Unit/RatePolicyTests.cs @@ -0,0 +1,115 @@ +using Esiur.Data.Types; +using Esiur.Protocol; +using Esiur.Resource; +using Esiur.Security.Authority; +using Esiur.Security.Permissions; +using Esiur.Security.RateLimiting; +using System.Reflection; + +namespace Esiur.Tests.Unit; + +public class RatePolicyTests +{ + [Fact] + public void RateControlAttribute_IsCapturedForFunctionsAndProperties() + { + var warehouse = new Warehouse(); + var method = typeof(RateFixture).GetMethod(nameof(RateFixture.Call))!; + var property = typeof(RateFixture).GetProperty(nameof(RateFixture.Value))!; + + var functionDefinition = FunctionDef.MakeFunctionDef( + warehouse, typeof(RateFixture), method, 0, method.Name, new TypeDef()); + var propertyDefinition = PropertyDef.MakePropertyDef( + warehouse, typeof(RateFixture), property, property.Name, 0, new TypeDef()); + + Assert.Equal("standard-call", functionDefinition.RatePolicyName); + Assert.Equal("standard-set", propertyDefinition.RatePolicyName); + } + + [Fact] + public void Warehouse_RegistersPoliciesByTheirConfiguredName() + { + var warehouse = new Warehouse(); + var policy = new DenyPolicy { Name = "deny" }; + + warehouse.AddRatePolicy(policy); + + Assert.Same(policy, warehouse.TryGetRatePolicy("deny")); + Assert.Throws(() => warehouse.AddRatePolicy(policy)); + Assert.True(warehouse.RemoveRatePolicy("deny")); + Assert.Null(warehouse.TryGetRatePolicy("deny")); + } + + [Fact] + public void ContextFreePolicies_AreSupported() + { + var policy = new DenyPolicy(); + Assert.Equal(Ruling.Denied, policy.Applicable(CreateContext())); + } + + [Fact] + public void BurstPolicy_AllowsBurstQueuesOverflowAndThenDenies() + { + var policy = new BurstRatePolicy("standard-call") + { + PermitLimit = 1, + Period = TimeSpan.FromSeconds(1), + BurstLimit = 1, + QueueLimit = 1, + }; + + var immediateOne = CreateContext(); + var immediateTwo = CreateContext(immediateOne.Connection); + var queued = CreateContext(immediateOne.Connection); + var denied = CreateContext(immediateOne.Connection); + + Assert.Equal(Ruling.Allowed, policy.Applicable(immediateOne)); + Assert.Equal(Ruling.Allowed, policy.Applicable(immediateTwo)); + Assert.Equal(Ruling.Allowed, policy.Applicable(queued)); + Assert.True(queued.Delay > TimeSpan.Zero); + Assert.Equal(Ruling.Denied, policy.Applicable(denied)); + } + + [Fact] + public void WarehouseConfiguration_IsIsolatedPerWarehouse() + { + var configured = new Warehouse(new WarehouseConfiguration + { + RateControl = new RateControlConfiguration + { + DenialsBeforeConnectionBlock = 2, + DenialWindow = TimeSpan.FromSeconds(10), + ConnectionBlockDelay = TimeSpan.Zero, + }, + }); + var defaults = new Warehouse(); + + Assert.Equal(2, configured.Configuration.RateControl.DenialsBeforeConnectionBlock); + Assert.Equal(5, defaults.Configuration.RateControl.DenialsBeforeConnectionBlock); + } + + static RateControlContext CreateContext(EpConnection? connection = null) + => new( + new Warehouse(), + connection ?? new EpConnection(), + new Session(), + null, + new FunctionDef { Name = "Call" }, + ActionType.Execute); + + sealed class DenyPolicy : RatePolicy + { + public override Ruling Applicable() => Ruling.Denied; + } + + sealed class RateFixture + { + [RateControl("standard-call")] + public void Call() + { + } + + [RateControl("standard-set")] + public int Value { get; set; } + } +} diff --git a/Tests/Unit/TypeDefAttributeTests.cs b/Tests/Unit/TypeDefAttributeTests.cs new file mode 100644 index 0000000..b5bb423 --- /dev/null +++ b/Tests/Unit/TypeDefAttributeTests.cs @@ -0,0 +1,177 @@ +using Esiur.Core; +using Esiur.Data; +using Esiur.Data.Types; +using Esiur.Resource; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; + +namespace Esiur.Tests.Unit; + +public class TypeDefAttributeTests +{ + [Fact] + public void FunctionStreamMode_IsInferredOrReadFromAttribute() + { + var pull = MakeFunction(nameof(StreamFixture.Pull)); + var push = MakeFunction(nameof(StreamFixture.Push)); + var explicitStream = MakeFunction(nameof(StreamFixture.Explicit)); + + Assert.Equal(StreamMode.Pull, pull.StreamMode); + Assert.Equal(TruIdentifier.Int32, pull.ReturnType.Identifier); + Assert.Equal(StreamMode.Push, push.StreamMode); + Assert.Equal(TruIdentifier.Int32, push.ReturnType.Identifier); + Assert.Equal(StreamMode.Push, explicitStream.StreamMode); + Assert.True(explicitStream.Pausable); + } + + [Theory] + [InlineData(nameof(StreamFixture.InvalidReturn))] + [InlineData(nameof(StreamFixture.InvalidPausablePull))] + [InlineData(nameof(StreamFixture.ConflictingMode))] + public void FunctionStreamMode_RejectsInvalidDeclarations(string methodName) + { + var exception = Assert.Throws(() => MakeFunction(methodName)); + Assert.Contains("stream", exception.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public void Attributes_AreIncludedInSerializedTypeDef() + { + var warehouse = new Warehouse(); + var definition = new LocalTypeDef(typeof(AttributedResource), warehouse); + var bytes = Codec.Compose(TypeDefInfo.FromTypeDef(definition), warehouse, null); + var (_, info) = Codec.ParseIndexedType(bytes, 0, warehouse); + + Assert.Equal("type usage", info.Usage); + Assert.Equal("type description", info.Description); + Assert.Equal("type example", info.Example); + Assert.Equal("tests", info.Category); + Assert.Equal("3.1", info.Since); + Assert.Equal("type", info.Annotations!["scope"]); + + var property = Assert.Single(info.Properties!, x => x.Name == "Value"); + var propertyFlags = (PropertyDefFlags)property.Flags; + Assert.True(propertyFlags.HasFlag(PropertyDefFlags.ReadOnly)); + Assert.True(propertyFlags.HasFlag(PropertyDefFlags.Volatile)); + Assert.True(propertyFlags.HasFlag(PropertyDefFlags.Historical)); + Assert.True(propertyFlags.HasFlag(PropertyDefFlags.Deprecated)); + Assert.Equal(OrderingControl.LatestOnly, property.OrderingControl); + Assert.Equal(5, Convert.ToInt32(property.DefaultValue)); + Assert.Equal("property description", property.Description); + Assert.Equal("property usage", property.Usage); + Assert.Equal(2, property.Examples!.Count); + Assert.Equal(new[] { "state", "sample" }, property.Tags); + Assert.Equal("units", property.Unit); + Assert.Equal(0, Convert.ToInt32(property.Minimum)); + Assert.Equal(10, Convert.ToInt32(property.Maximum)); + Assert.Equal(new[] { 1, 5 }, property.AllowedValues!.Select(Convert.ToInt32)); + Assert.Equal("^[0-9]+$", property.Pattern); + Assert.Equal("integer", property.Format); + Assert.Equal("property warning", Assert.Single(property.Warnings!)); + Assert.Equal((byte)7, Assert.Single(property.RelatedMembers!)); + Assert.Equal("use NewValue", property.DeprecationMessage); + Assert.Equal("property", property.Annotations!["scope"]); + + var function = Assert.Single(info.Functions!, x => x.Name == nameof(AttributedResource.Watch)); + var functionFlags = (FunctionDefFlags)function.Flags; + Assert.True(functionFlags.HasFlag(FunctionDefFlags.ReadOnly)); + Assert.True(functionFlags.HasFlag(FunctionDefFlags.Idempotent)); + Assert.True(functionFlags.HasFlag(FunctionDefFlags.Cancellable)); + Assert.True(functionFlags.HasFlag(FunctionDefFlags.Pausable)); + Assert.Equal(StreamMode.Push, function.StreamMode); + Assert.Equal("ready", Assert.Single(function.Preconditions!)); + Assert.Equal("complete", Assert.Single(function.Postconditions!)); + Assert.Equal(OperationEffects.EmitsEvents, function.Effects); + + var eventInfo = Assert.Single(info.Events!, x => x.Name == nameof(AttributedResource.Changed)); + var eventFlags = (EventDefFlags)eventInfo.Flags; + Assert.True(eventFlags.HasFlag(EventDefFlags.AutoDelivered)); + Assert.True(eventFlags.HasFlag(EventDefFlags.Historical)); + Assert.Equal(OrderingControl.Relaxed, eventInfo.OrderingControl); + + var constant = Assert.Single(info.Constants!, x => x.Name == nameof(AttributedResource.Limit)); + Assert.Equal("constant description", constant.Description); + Assert.Equal("constant usage", constant.Usage); + Assert.Equal(10, Convert.ToInt32(constant.Value)); + } + + private static FunctionDef MakeFunction(string name) + { + var method = typeof(StreamFixture).GetMethod(name, BindingFlags.Public | BindingFlags.Instance)!; + return FunctionDef.MakeFunctionDef( + Warehouse.Default, typeof(StreamFixture), method, 0, name, new TypeDef()); + } + + private sealed class StreamFixture + { + public IAsyncEnumerable Pull() => throw new NotSupportedException(); + + public IEnumerable Push() => Array.Empty(); + + [Stream(StreamMode.Push, Pausable = true)] + public AsyncReply Explicit() => new(1); + + [Stream] + public int InvalidReturn() => 1; + + [Stream(StreamMode.Pull, Pausable = true)] + public AsyncReply InvalidPausablePull() => new(1); + + [Stream(StreamMode.Push)] + public IAsyncEnumerable ConflictingMode() => throw new NotSupportedException(); + } + + [Export] + [Usage("type usage")] + [Description("type description")] + [Example("type example")] + [Category("tests")] + [Since("3.1")] + [Annotation("scope", "type")] + private sealed class AttributedResource : Esiur.Resource.Resource + { + [Description("constant description")] + [Usage("constant usage")] + [Example(10)] + [Annotation("scope", "constant")] + public const int Limit = 10; + + [ReadOnly] + [Volatile] + [Historical] + [Ordering(OrderingControl.LatestOnly)] + [System.ComponentModel.DefaultValue(5)] + [Description("property description")] + [Usage("property usage")] + [Example(1)] + [Example(5)] + [Tags("state", "sample")] + [Unit("units")] + [Minimum(0)] + [Maximum(10)] + [AllowedValue(1)] + [AllowedValue(5)] + [Pattern("^[0-9]+$")] + [Format("integer")] + [Warning("property warning")] + [RelatedMembers(7)] + [Obsolete("use NewValue")] + [Annotation("scope", "property")] + public int Value { get; set; } + + [Stream(StreamMode.Push, Pausable = true)] + [ReadOnly] + [Idempotent] + [Cancellable] + [Precondition("ready")] + [Postcondition("complete")] + [Effects(OperationEffects.EmitsEvents)] + public AsyncReply Watch() => new(1); + + [AutoDelivery] + [Historical] + [Ordering(OrderingControl.Relaxed)] + public event ResourceEventHandler? Changed; + } +}