Recently I had to implement a Windows Service that connects to the Shopify REST API and pulls essentially everything new into some SQL Server database, on a hourly basis; I've also implemented a small console app off this code, that can query the REST API from command-line.
The Shopify REST API is rate-limited (using the "leaky bucket" method), obviously, so I can't just send all requests involved all at once. The rate limit is per API key, per REST endpoint. When a response is received, the headers include the throttling metadata.
So in order to adhere to best practices, I implemented a RequestDispatcher class whose role is to... dispatch requests, and have a way to ensure all responses get handled before the dispatcher gets disposed: clients can do that by invoking the WaitForCompletion method, which blocks the current thread until the both CountdownEvent counters reach 0.
Clients of this class simply Enqueue an IDispatchedRequest, and the dispatcher decides when the request is actually issued, using a timer that I've configured to tick 10 times per second (ticks are skipped if throttling metadata warrants waiting a bit).
This is what an IDispatchedRequest looks like (that's the non-generic version - there are other variants extending it):
public interface IDispatchedRequest
{
int Retries { get; set; }
string ResourceName { get; }
Func<Task> Request { get; }
Func<object,Task> OnResponse { get; }
}
There are various implementations of that interface, notably DispatchedRequest, DispatchedEntityRequest, DispatchedPostRequest, and DispatchedChildEntititesRequest. In every case, the idea is to encapsulate a Func delegate that returns a Task<Response<T>> (I'm using RestEase to deal with all the JSON deserialization, so that's a RestEase.Response<T>), and a Func delegate that takes the response and handles it.
I'm pretty happy with the implementation, I find it looks pretty clean in general, but the part that handles exceptions is rather chaotic... yet everything works as intended as far as I can tell.
public class RequestDispatcher : IDisposable
{
private readonly ILogger _logger;
private readonly INotifierService _notifier;
private readonly int _appId;
private readonly string _connectionString;
public static readonly int ShopifyDefaultBucketCapacity = Settings.Default.ShopifyDefaultBucketCapacity;
public static readonly int ShopifyPlusBucketCapacity = Settings.Default.ShopifyPlusBucketCapacity;
private readonly double TicksPerSecond;
private readonly double _cutoff;
private CountdownEvent _requestWaitHandle;
private CountdownEvent _responseWaitHandle;
private readonly Queue<IDispatchedRequest> _requestQueue
= new Queue<IDispatchedRequest>();
private readonly System.Timers.Timer _timer; // would resolve to System.Threading.Timer unqualified
private readonly HashSet<IDispatchedRequest> _pendingHandlers
= new HashSet<IDispatchedRequest>();
private readonly ConcurrentDictionary<string, ShopifyThrottleInfo> _throttleInfo
= new ConcurrentDictionary<string, ShopifyThrottleInfo>();
private readonly bool _verbose;
public RequestDispatcher(INotifierService notifier, int appId, string connectionString, double waitCutoff = 0.6, double? ticksPerSecond = null, bool outputJson = false)
{
_logger = LogManager.GetCurrentClassLogger();
TicksPerSecond = ticksPerSecond ?? Settings.Default.DispatcherTicksPerSecond;
_notifier = notifier;
_appId = appId;
_connectionString = connectionString;
_cutoff = waitCutoff;
_verbose = outputJson;
_timer = new System.Timers.Timer(1000 / TicksPerSecond);
_timer.Elapsed += OnTick;
_timer.Start();
}
public int PendingHandlers => _pendingHandlers.Count;
public int PendingRequests => _requestQueue.Count;
public void Pause() => _timer.Stop();
public void Resume() => _timer.Start();
private void WaitForPendingRequests() => _requestWaitHandle?.Wait();
private void WaitForPendingHandlers() => _responseWaitHandle?.Wait();
public void WaitForCompletion()
{
WaitForPendingRequests();
WaitForPendingHandlers();
}
public void Enqueue<TAggregate, TWrapper, TEntity>(IDispatchedRequest<TAggregate, TWrapper, TEntity> request)
where TWrapper : class, IEntityWrapper<TEntity>
where TAggregate : class, IAggregateEntity<TEntity>
where TEntity : class
{
ThrowIfDisposed();
if (request == null) { throw new ArgumentNullException(); }
_requestQueue.Enqueue(request);
_pendingHandlers.Add(request);
IncrementWaitHandle(ref _requestWaitHandle);
IncrementWaitHandle(ref _responseWaitHandle);
_logger.Debug($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Enqueued {request.ResourceName} request (hashcode {request.GetHashCode()}); {_requestQueue.Count} requests enqueued");
}
public void Enqueue<TWrapper, TEntity>(IDispatchedRequest<TWrapper, TEntity> request)
where TWrapper : class, IEntityWrapper<TEntity>
where TEntity : class
{
ThrowIfDisposed();
if (request == null) { throw new ArgumentNullException(); }
_requestQueue.Enqueue(request);
_pendingHandlers.Add(request);
IncrementWaitHandle(ref _requestWaitHandle);
IncrementWaitHandle(ref _responseWaitHandle);
_logger.Debug($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Enqueued {request.ResourceName} request (hashcode {request.GetHashCode()}); {_requestQueue.Count} requests enqueued");
}
public void Enqueue(IDispatchedRequest request)
{
ThrowIfDisposed();
if (request == null) { throw new ArgumentNullException(); }
_requestQueue.Enqueue(request);
_pendingHandlers.Add(request);
IncrementWaitHandle(ref _requestWaitHandle);
IncrementWaitHandle(ref _responseWaitHandle);
_logger.Debug($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Enqueued {request.ResourceName} request (hashcode {request.GetHashCode()}); {_requestQueue.Count} requests enqueued");
}
private void IncrementWaitHandle(ref CountdownEvent handle)
{
if (handle == null)
{
handle = new CountdownEvent(1);
}
else if (handle.IsSet)
{
handle.Dispose();
handle = new CountdownEvent(1);
}
else
{
handle.AddCount();
}
}
private int _skippedTicks = 0;
private bool ShouldSkip(string resource)
{
ShopifyThrottleInfo info;
if (_throttleInfo.TryGetValue(resource, out info))
{
if (info.RetryAfter.TotalSeconds > 0)
{
// any premature request will be denied. insisting may get the API key disabled.
return _skippedTicks < TicksPerSecond * info.RetryAfter.TotalSeconds;
}
// skip {TicksPerSeconds} ticks (1 second) if {_cutoff} % capacity is hit for this endpoint.
return info.PercentCapacity > _cutoff && _skippedTicks < TicksPerSecond;
}
return false;
}
private void OnTick(object sender, ElapsedEventArgs e)
{
ThrowIfDisposed();
if (_requestQueue.Count == 0)
{
return;
}
var next = _requestQueue.Peek();
if (next == null)
{
_logger.Warn("Dequeued a null reference. Next up is a null request?? Dequeuing...");
_requestQueue.Dequeue();
return;
}
if (ShouldSkip(next.ResourceName))
{
_skippedTicks++;
_logger.Warn($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Skipped {_skippedTicks} dispatcher tick{(_skippedTicks > 1 ? "s" : string.Empty)}.");
return;
}
#pragma warning disable CS4014
// Because this call is not awaited, execution of the current method continues before the call is completed.
// ...that's the whole point...
IssueNextRequest();
#pragma warning restore CS4014
}
private async Task IssueNextRequest()
{
ThrowIfDisposed();
_skippedTicks = 0;
var request = GetNextRequest();
Debug.Assert(request != null);
dynamic response = null;
try
{
// we can't bind T at compile-time for the generic Task<Response<T>>
// so we bind it at run-time through a dynamic dispatch:
var task = ((dynamic)request).Request.Invoke();
response = task.Result; // response is Response<T>, where T is usually TAggregate, or TEntity otherwise.
if (_verbose)
{
Console.WriteLine(response.StringContent); // dynamic dispatch here
}
_logger.Info($"Successfully retrieved response from dispatched {request.ResourceName} request (hashcode {request.GetHashCode()}).");
if (response != null)
{
await HandleResponse(request, response);
}
else
{
_logger.Warn($"respone (request hashcode {request.GetHashCode()}) is null.");
}
}
catch (AggregateException e)
{
if (!HandleApiException(request, e))
{
throw;
}
}
finally
{
_pendingHandlers.Remove(request);
_responseWaitHandle.Signal();
}
}
private async Task HandleResponse(IDispatchedRequest request, dynamic response)
{
ShopifyThrottleInfo throttleInfo = ResponseExtensions.GetThrottleInfo(response, request.ResourceName);
_throttleInfo.AddOrUpdate(request.ResourceName, throttleInfo, (key, value) => throttleInfo);
_logger.Debug($"Updated throttling info for {request.ResourceName}: {throttleInfo.Value}/{throttleInfo.Capacity} ({throttleInfo.PercentCapacity * 100}%)");
try
{
_logger.Trace($"Handling response from dispatched {request.ResourceName} request (hashcode: {request.GetHashCode()})");
await request.OnResponse(response);
}
finally
{
_logger.Trace($"Response of {request.ResourceName} request (hashcode: {request.GetHashCode()}) has been handled.");
_logger.Debug($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Pending requests: {_requestQueue.Count}; pending handlers: {_pendingHandlers.Count}; wait handle counter: {_responseWaitHandle.CurrentCount}");
}
}
private bool HandleApiException(IDispatchedRequest request, AggregateException e)
{
var handled = true; // caller will rethrow exception if method returns false.
var apiExceptions = e.InnerExceptions.OfType<ApiException>();
foreach (var exception in apiExceptions)
{
switch ((int)exception.StatusCode)
{
case 401: // unauthorized (verify authentication)
case 403: // forbidden (verify permissions)
handled = false;
break;
case 422: // unprocessable entity (e.g. posted fulfillment but line item was already fulfilled *LOG THIS*)
_logger.Warn(exception);
break;
case 404: // not found (e.g. ShopifyID refers to non-existing resource *LOG THIS*)
// also returned by "fulfillments" endpoint (POST) if the specified location isn't active.
_logger.Warn(exception);
var postRequest = request as DispatchedPostRequest<FulfillmentWrapper, Fulfillment, FulfillmentPosting>;
if (postRequest != null)
{
NotifyFailedFulfillmentPosting(exception, postRequest);
}
break;
case 429: // too many requests
// if this ApiException was thrown, tweak the throttling knobs...
_logger.Error(exception);
if (request.Retries < 3)
{
// todo log re-enqueued request
Task.Delay(TimeSpan.FromSeconds(10))
.ContinueWith(_ =>
{
request.Retries++;
//Enqueue(request); // compiler can't resolve overload w/dynamic dispatch here
_requestQueue.Enqueue(request);
IncrementWaitHandle(ref _requestWaitHandle);
});
}
else
{
_notifier.Enqueue($"Failed to dispatch request {request.Retries} times (hashcode {request.GetHashCode()}). Request is aborted, data will be missing. Review logs for details.");
}
break;
}
_logger.Error(e);
}
return handled;
}
private IDispatchedRequest GetNextRequest()
{
var request = (IDispatchedRequest)null;
try
{
request = _requestQueue.Dequeue();
_requestWaitHandle.Signal();
_logger.Debug($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Dequeued {request.ResourceName} request (hashcode: {request.GetHashCode()})");
return request;
}
catch (Exception exception)
{
_logger.Error(exception);
}
return null;
}
private bool _disposed = false;
public void Dispose()
{
_timer.Elapsed -= OnTick;
_timer.Dispose();
_logger.Info($"{DateTime.UtcNow.ToString("HH:mm:ss.fff")} Disposing wait handle with {_requestQueue.Count} requests queued and {_pendingHandlers.Count} pending handlers.");
_requestQueue.Clear();
_requestWaitHandle?.Dispose();
_pendingHandlers.Clear();
_responseWaitHandle?.Dispose();
_disposed = true;
}
private void ThrowIfDisposed()
{
if (_disposed)
{
throw new ObjectDisposedException(nameof(RequestDispatcher));
}
}
}
Note that I've omitted the failed fulfillment posting stuff, which arguably belongs to another class.
How should I clean up the exception-handling code, and does anything else look like it could be improved?
