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?