This is a request-response model over sockets/websockets (like HTTP) where you technically match request id to response id and return the response. In addition to that, there is timeout in case that message is never matched. It does its job but the implementation is not disposing the CancellationTokenSource and TaskCompletionSource. It may even be done in a better way, so here is why I want a code review.
Example usage
var requestManager = new RequestManager();
var input = JsonSerializer.Serialize(new Test { Message = "Hello World" });
var pending = requestManager.AddPendingRequest(x =>
{
var text = x.Deserialize<Test>();
return text?.Message == "Hello World";
}, TimeSpan.FromSeconds(5));
await Task.Delay(4000);
var jsonElement = JsonSerializer.Deserialize<JsonElement>(input);
if (requestManager.TryMatchRequest(jsonElement))
{
Console.WriteLine("Found");
}
public class Test
{
public string Message { get; set; }
}
Code to review
public class PendingRequest
{
private readonly CancellationTokenSource _cts;
public PendingRequest(Func<JsonElement, bool> handler, TimeSpan timeout)
{
Handler = handler;
Event = new AsyncResetEvent(false, false);
Timeout = timeout;
_cts = new CancellationTokenSource(timeout);
_cts.Token.Register(Fail, false);
}
public Func<JsonElement, bool> Handler { get; }
public JsonElement? Result { get; private set; }
public bool Completed { get; private set; }
public AsyncResetEvent Event { get; }
public TimeSpan Timeout { get; }
public bool CheckData(JsonElement data)
{
if (Handler(data))
{
Result = data;
Completed = true;
Event.Set();
return true;
}
return false;
}
public void Fail()
{
Completed = true;
Event.Set();
}
}
public class RequestManager
{
private readonly IList<PendingRequest> _pendingRequests = new List<PendingRequest>();
public PendingRequest AddPendingRequest(Func<JsonElement, bool> handler, TimeSpan timeout)
{
var pending = new PendingRequest(handler, timeout);
lock (_pendingRequests)
{
_pendingRequests.Add(pending);
}
return pending;
}
public void FailAllPendingRequests()
{
lock (_pendingRequests)
{
foreach (var pendingRequest in _pendingRequests.ToList())
{
pendingRequest.Fail();
_pendingRequests.Remove(pendingRequest);
}
}
}
public IList<PendingRequest> GetAllPendingRequests()
{
lock (_pendingRequests)
{
return _pendingRequests;
}
}
public bool TryMatchRequest(JsonElement tokenData)
{
lock (_pendingRequests)
{
foreach (var request in _pendingRequests)
{
if (!request.CheckData(tokenData))
{
continue;
}
_pendingRequests.Remove(request);
return true;
}
}
return false;
}
public void RemoveTimedOutRequests()
{
PendingRequest[] requests;
lock (_pendingRequests)
{
requests = _pendingRequests.ToArray();
}
foreach (var request in requests.Where(r => r.Completed))
{
lock (_pendingRequests)
{
_pendingRequests.Remove(request);
}
}
}
}
/// <summary>
/// Async auto reset based on Stephen Toub`s implementation
/// https://devblogs.microsoft.com/pfxteam/building-async-coordination-primitives-part-2-asyncautoresetevent/
/// </summary>
public sealed class AsyncResetEvent : IDisposable
{
private static readonly Task<bool> Completed = Task.FromResult(true);
private readonly Queue<TaskCompletionSource<bool>> _waits = new();
private readonly bool _reset;
private bool _signaled;
/// <summary>
/// New AsyncResetEvent
/// </summary>
/// <param name="initialState"></param>
/// <param name="reset"></param>
public AsyncResetEvent(bool initialState = false, bool reset = true)
{
_signaled = initialState;
_reset = reset;
}
/// <summary>
/// Dispose
/// </summary>
public void Dispose()
{
_waits.Clear();
}
/// <summary>
/// Wait for the AutoResetEvent to be set
/// </summary>
/// <returns></returns>
public Task<bool> WaitAsync(TimeSpan? timeout = null)
{
lock (_waits)
{
if (_signaled)
{
if (_reset)
{
_signaled = false;
}
return Completed;
}
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
if (timeout != null)
{
var cancellationSource = new CancellationTokenSource(timeout.Value);
cancellationSource.Token.Register(() =>
{
tcs.TrySetResult(false);
}, false);
}
_waits.Enqueue(tcs);
return tcs.Task;
}
}
/// <summary>
/// Signal a waiter
/// </summary>
public void Set()
{
lock (_waits)
{
if (!_reset)
{
// Act as ManualResetEvent. Once set keep it signaled and signal everyone who is waiting
_signaled = true;
while (_waits.Count > 0)
{
var toRelease = _waits.Dequeue();
toRelease.TrySetResult(true);
}
}
else
{
// Act as AutoResetEvent. When set signal 1 waiter
if (_waits.Count > 0)
{
var toRelease = _waits.Dequeue();
toRelease.TrySetResult(true);
}
else if (!_signaled)
{
_signaled = true;
}
}
}
}
}
```
