Rikitav/Telegrator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 13 Wiki Activity

* Added mising XML summaries

Browse Source 
* Obsolete code comments cleanup
This commit is contained in:
Rikitav
2025-08-02 03:33:35 +04:00
parent 02e23eefdd
commit 7d1ce6ea22
4 changed files with 58 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Telegrator/Polling/LimitedDictionary.cs
+26
View File
@@ -3,12 +3,21 @@ using System.Collections.Concurrent;
namespace Telegrator.Polling
{
/// <summary>
/// Represents a dictionayr with limited number of slots, if trying to overflow, will block calling thread until one of slots will be free
/// </summary>
/// <typeparam name="TKey"></typeparam>
/// <typeparam name="TValue"></typeparam>
public class LimitedDictionary<TKey, TValue> : IEnumerable<KeyValuePair<TKey, TValue>>, IDisposable
{
private readonly int? _maximum;
private readonly SemaphoreSlim _semaphore = null!;
private readonly ConcurrentDictionary<TKey, TValue> _dict = [];
/// <summary>
/// Initializes new instance of <see cref="LimitedDictionary{TKey, TValue}"/>
/// </summary>
/// <param name="maximum"></param>
public LimitedDictionary(int? maximum)
{
_maximum = maximum;
@@ -19,6 +28,14 @@ namespace Telegrator.Polling
}
}
/// <summary>
/// Tries to add new element to dictioanry.
/// If all slots are occupied, blocks calling thread.
/// </summary>
/// <param name="key"></param>
/// <param name="value"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async Task<bool> Add(TKey key, TValue value, CancellationToken cancellationToken)
{
if (_semaphore != null)
@@ -27,6 +44,13 @@ namespace Telegrator.Polling
return _dict.TryAdd(key, value);
}
/// <summary>
/// Tries to remove element from dictionay.
/// Frees slot on success.
/// </summary>
/// <param name="key"></param>
/// <param name="result"></param>
/// <returns></returns>
public bool Remove(TKey key, out TValue result)
{
if (_dict.TryRemove(key, out result))
@@ -38,8 +62,10 @@ namespace Telegrator.Polling
return false;
}
/// <inheritdoc/>
public IEnumerator<KeyValuePair<TKey, TValue>> GetEnumerator() => _dict.GetEnumerator();
/// <inheritdoc/>
IEnumerator IEnumerable.GetEnumerator() => _dict.GetEnumerator();
/// <inheritdoc/>
Telegrator/Polling/LimitedQueue.cs
+2
View File
@@ -2,6 +2,7 @@
namespace Telegrator.Polling
{
/*
public class LimitedQueue<T>
{
private readonly int? _maximum;
@@ -39,4 +40,5 @@ namespace Telegrator.Polling
return false;
}
}
*/
}
Telegrator/Polling/UpdateHandlersPool.cs
+1 -166
View File
@@ -1,5 +1,4 @@
using System.Collections.Concurrent;
using Telegrator.Handlers;
using Telegrator.Handlers;
using Telegrator.MadiatorCore;
using Telegrator.MadiatorCore.Descriptors;
@@ -16,34 +15,11 @@ namespace Telegrator.Polling
/// </summary>
protected object SyncObj = new object();
/*
/// <summary>
/// Event that signals when awaiting handlers are queued.
/// </summary>
protected ManualResetEventSlim AwaitingHandlersQueuedEvent = null!;
/// <summary>
/// Semaphore for controlling the number of concurrently executing handlers.
/// </summary>
protected SemaphoreSlim ExecutingHandlersSemaphore = null!;
/// <summary>
/// Queue for storing awaiting handlers.
/// </summary>
protected readonly ConcurrentQueue<DescribedHandlerInfo> AwaitingHandlersQueue = [];
/// <summary>
/// Dictionary for tracking currently executing handlers.
/// </summary>
protected readonly ConcurrentDictionary<HandlerLifetimeToken, Task> ExecutingHandlersPool = [];
*/
//protected readonly ConcurrentDictionary<Type, LimitedQueue<DescribedHandlerInfo>> AwaitingHandlersQueue;
//protected readonly LimitedDictionary<HandlerLifetimeToken, Task> ExecutingHandlersPool;
protected SemaphoreSlim ExecutingHandlersSemaphore = null!;
/// <summary>
/// The bot configuration options.
/// </summary>
@@ -74,26 +50,16 @@ namespace Telegrator.Polling
{
Options = options;
GlobalCancellationToken = globalCancellationToken;
//AwaitingHandlersQueue = new ConcurrentDictionary<Type, LimitedQueue<DescribedHandlerInfo>>();
//ExecutingHandlersPool = new LimitedDictionary<HandlerLifetimeToken, Task>(options.MaximumParallelWorkingHandlers);
if (options.MaximumParallelWorkingHandlers != null)
{
ExecutingHandlersSemaphore = new SemaphoreSlim(options.MaximumParallelWorkingHandlers.Value);
//AwaitingHandlersQueuedEvent = new ManualResetEventSlim(false);
}
/*
if (Options.MaximumParallelWorkingHandlers != null)
HandlersCheckpoint();
*/
}
/// <inheritdoc/>
public async Task Enqueue(IEnumerable<DescribedHandlerInfo> handlers)
{
//handlers.ForEach(Enqueue);
Result? lastResult = null;
foreach (DescribedHandlerInfo handlerInfo in handlers)
{
@@ -123,137 +89,6 @@ namespace Telegrator.Polling
break;
}
}
/*
/// <inheritdoc/>
public void Enqueue(DescribedHandlerInfo handlerInfo)
{
throw new NotImplementedException();
if (Options.MaximumParallelWorkingHandlers == null)
{
Task.Run(async () => await ExecuteHandlerWrapper(handlerInfo));
return;
}
lock (SyncObj)
{
AwaitingHandlersQueue.Enqueue(handlerInfo);
HandlerEnqueued?.Invoke(handlerInfo);
AwaitingHandlersQueuedEvent.Set();
}
}
/// <inheritdoc/>
public void Dequeue(HandlerLifetimeToken token)
{
if (Options.MaximumParallelWorkingHandlers == null)
return;
lock (SyncObj)
{
ExecutingHandlersPool.TryRemove(token, out _);
ExecutingHandlersSemaphore.Release(1);
}
}
*/
/*
/// <summary>
/// Main checkpoint method that manages handler execution in a loop.
/// Continuously processes queued handlers while respecting concurrency limits.
/// </summary>
protected virtual async void HandlersCheckpoint()
{
await Task.Yield();
while (!GlobalCancellationToken.IsCancellationRequested)
{
if (!CanEnqueueHandler())
{
await ExecutingHandlersSemaphore.WaitAsync(GlobalCancellationToken);
if (!CanEnqueueHandler())
continue;
}
if (!TryDequeueHandler(out DescribedHandlerInfo? enqueuedHandler))
{
AwaitingHandlersQueuedEvent.Reset();
AwaitingHandlersQueuedEvent.Wait(GlobalCancellationToken);
if (!TryDequeueHandler(out enqueuedHandler))
continue;
}
if (enqueuedHandler == null)
continue;
ExecuteHandler(enqueuedHandler);
}
}
/// <summary>
/// Executes a handler by creating a lifetime token and tracking the execution.
/// </summary>
/// <param name="enqueuedHandler">The handler to execute.</param>
protected virtual void ExecuteHandler(DescribedHandlerInfo enqueuedHandler)
{
HandlerLifetimeToken lifetimeToken = enqueuedHandler.HandlerLifetime;
lifetimeToken.OnLifetimeEnded += Dequeue;
Task executingHandler = ExecuteHandlerWrapper(enqueuedHandler);
lock (SyncObj)
ExecutingHandlersPool.TryAdd(lifetimeToken, executingHandler);
HandlerExecuting?.Invoke(enqueuedHandler);
}
/// <summary>
/// Wrapper method that executes a handler and handles exceptions.
/// </summary>
/// <param name="enqueuedHandler">The handler to execute.</param>
/// <returns>A task representing the asynchronous execution.</returns>
/// <exception cref="HandlerFaultedException">Thrown when the handler execution fails.</exception>
protected virtual async Task ExecuteHandlerWrapper(DescribedHandlerInfo enqueuedHandler)
{
try
{
await enqueuedHandler.Execute(GlobalCancellationToken);
}
catch (OperationCanceledException)
{
return;
}
catch (Exception ex)
{
throw new HandlerFaultedException(enqueuedHandler, ex);
}
}
/// <summary>
/// Checks if a new handler can be enqueued based on the current execution count.
/// </summary>
/// <returns>True if a new handler can be enqueued; otherwise, false.</returns>
protected virtual bool CanEnqueueHandler()
{
lock (SyncObj)
{
return ExecutingHandlersPool.Count < Options.MaximumParallelWorkingHandlers;
}
}
/// <summary>
/// Attempts to dequeue a handler from the awaiting queue.
/// </summary>
/// <param name="enqueuedHandler">The dequeued handler, if successful.</param>
/// <returns>True if a handler was successfully dequeued; otherwise, false.</returns>
protected virtual bool TryDequeueHandler(out DescribedHandlerInfo? enqueuedHandler)
{
lock (SyncObj)
{
return AwaitingHandlersQueue.TryDequeue(out enqueuedHandler);
}
}
*/
/// <summary>
/// Disposes of the handlers pool and releases all resources.