add amazon sqs support #85
Conversation
# Conflicts: # src/Foundatio.AWS/Foundatio.AWS.csproj # test/Foundatio.AWS.Tests/Foundatio.AWS.Tests.csproj
# Conflicts: # src/Foundatio.AWS/Foundatio.AWS.csproj # test/Foundatio.AWS.Tests/Foundatio.AWS.Tests.csproj
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
| using Foundatio.Utility; | ||
| using Nito.AsyncEx; | ||
|
|
||
| namespace Foundatio.AWS.Queues { |
There was a problem hiding this comment.
All our other queues are in the Foundatio.Queues namespace
| if (!String.IsNullOrEmpty(_queueUrl)) | ||
| return; | ||
|
|
||
| using (await _lock.LockAsync(cancellationToken).ConfigureAwait(false)) { |
There was a problem hiding this comment.
We have an AnyContext() extension method for ConfigureAwait(false)
There was a problem hiding this comment.
Looks like this one was still missed :)
There was a problem hiding this comment.
That one is not supported, it returns something not supported by the extension method
There was a problem hiding this comment.
We should have an extension method for that, maybe it's missing from the project?
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
| MessageBody = await _serializer.SerializeToStringAsync(data).ConfigureAwait(false), | ||
| }; | ||
|
|
||
| var response = await _client.Value.SendMessageAsync(message).ConfigureAwait(false); |
There was a problem hiding this comment.
Should we be retrying this? We retry in half of our current implementations (curious what your thoughts are) https://github.com/exceptionless/Foundatio/blob/master/src/Foundatio.Redis/Queues/RedisQueue.cs#L201
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
| if (!await OnEnqueuingAsync(data).ConfigureAwait(false)) | ||
| return null; | ||
|
|
||
| Interlocked.Increment(ref _enqueuedCount); |
There was a problem hiding this comment.
In our other clients we do this after we publish the message.
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
| } | ||
|
|
||
| protected override async Task<IQueueEntry<T>> DequeueImplAsync(CancellationToken cancellationToken) { | ||
| var visibilityTimeout = Convert.ToInt32(_workItemTimeout.TotalSeconds); |
There was a problem hiding this comment.
It might be better to do conversions in the constructor instead of every time we dequeue it
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
| try { | ||
| response = await _client.Value.ReceiveMessageAsync(request, cancel).ConfigureAwait(false); | ||
| } | ||
| catch (TaskCanceledException) { |
There was a problem hiding this comment.
@StephenCleary recommended we catch the base OperationCanceledException exception for all cancellations.
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
|
|
||
| protected override void StartWorkingImpl(Func<IQueueEntry<T>, CancellationToken, Task> handler, bool autoComplete, CancellationToken cancellationToken) { | ||
| Task.Run(async () => { | ||
| while (!cancellationToken.IsCancellationRequested) { |
There was a problem hiding this comment.
we use the linked cancellation token here too, otherwise this task might never end (especially in unit tests) when the test is disposed
https://github.com/exceptionless/Foundatio/blob/master/src/Foundatio.Redis/Queues/RedisQueue.cs#L217
src/Foundatio.AWS/Queues/SQSQueue.cs
Outdated
| while (!cancellationToken.IsCancellationRequested) { | ||
| IQueueEntry<T> entry = null; | ||
| try { | ||
| entry = await DequeueImplAsync(cancellationToken).ConfigureAwait(false); |
There was a problem hiding this comment.
We seem to be breaking this apart into multiple try catches and checking the token to save work
https://github.com/exceptionless/Foundatio/blob/master/src/Foundatio.Redis/Queues/RedisQueue.cs#L226
https://github.com/exceptionless/Foundatio/blob/master/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs#L185
Is there any chance you could update the redis implementation (first try catch block should be catching OperationCanceledException)?
| <ProjectReference Include="..\Foundatio.Tests\Foundatio.Tests.csproj" /> | ||
| </ItemGroup> | ||
| <ItemGroup> | ||
| <PackageReference Include="log4net" Version="2.0.8" /> |
There was a problem hiding this comment.
These references are brought into automatically via our tests.props
There was a problem hiding this comment.
I don't think you meant to bring in log4net
There was a problem hiding this comment.
aws sdk looks for it via reflection. i get false errors if its missing.
| @@ -28,6 +28,10 @@ public abstract class QueueTestBase : TestWithLoggingBase, IDisposable { | |||
| return null; | |||
| } | |||
|
|
|||
| protected virtual Task CloseQueue(IQueue<SimpleWorkItem> queue) { | |||
There was a problem hiding this comment.
I wonder if our other azure queues and rabbit queues are being cleaned up, probably not :. I'm trying to think in the tests where we create multiple queue instances if they would need to be cleaned up or not. I almost think they wouldn't as they should all have the same queue name but different instance ids.
| // sqs doesn't support already canceled token, change timeout and token for sqs pattern | ||
| var waitTimeout = cancellationToken.IsCancellationRequested ? 0 : _readQueueTimeoutSeconds; | ||
|
|
||
| var cancel = cancellationToken.IsCancellationRequested |
There was a problem hiding this comment.
I don't see this being used anywhere
| _client.Value.Dispose(); | ||
|
|
||
| base.Dispose(); | ||
| } |
There was a problem hiding this comment.
The disposed cancellation token isn't being cancelled here.
|
Looks good to me!!! Thanks for the contrib |
|
I think the tests fail on this because it won't give it access to the secrets for SQS access. |
No description provided.