New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add AMQP retry mechanism and example projects #312
Merged
Aaronontheweb
merged 7 commits into
akkadotnet:dev
from
Arkatufus:AMQP_Add_retry_and_sample
Dec 18, 2020
Merged
Changes from all commits
Commits
Show all changes
7 commits
Select commit
Hold shift + click to select a range
dcd92ec
Add AMQP retry mechanism and example projects
Arkatufus 57134d1
Fix retry start index
Arkatufus e0fc457
Make a distinction between internally managed connection reference an…
Arkatufus 5059a11
Add better exception feedback on connection failure.
Arkatufus 4c7eb39
Make settings responsible for closing Session and Connection objects.
Arkatufus a3f06a7
Update examples to show usage of both QueueNamed and Address settings
Arkatufus b057321
Add IsPackable tag in example projects csproj files
Arkatufus File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
7 changes: 7 additions & 0 deletions
7
src/Amqp/Akka.Streams.Amqp.V1.Tests/Properties/launchSettings.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,7 @@ | ||
{ | ||
"profiles": { | ||
"Akka.Streams.Amqp.V1.Tests": { | ||
"commandName": "Project" | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,79 @@ | ||
using System; | ||
using System.Collections.Generic; | ||
using System.Text; | ||
using System.Threading.Tasks; | ||
using Akka.Serialization; | ||
using Amqp; | ||
using Amqp.Framing; | ||
using Amqp.Types; | ||
|
||
namespace Akka.Streams.Amqp.V1 | ||
{ | ||
public class AddressSinkSettings<T> : IAmqpSinkSettings<T> | ||
{ | ||
private readonly string _linkName; | ||
private readonly string _queueName; | ||
private readonly Serializer _serializer; | ||
private readonly Address _address; | ||
private readonly object _lock = new object(); | ||
|
||
private Connection _connection; | ||
private Session _session; | ||
|
||
public bool ManageConnection => true; | ||
|
||
public AddressSinkSettings( | ||
Address address, | ||
string linkName, | ||
string queueName, | ||
Serializer serializer) | ||
{ | ||
_address = address; | ||
_linkName = linkName; | ||
_queueName = queueName; | ||
_serializer = serializer; | ||
} | ||
|
||
public byte[] GetBytes(T obj) | ||
{ | ||
return _serializer.ToBinary(obj); | ||
} | ||
|
||
public void CloseConnection() | ||
{ | ||
_session?.Close(); | ||
_connection?.Close(); | ||
|
||
_session = null; | ||
_connection = null; | ||
} | ||
|
||
public async Task CloseConnectionAsync() | ||
{ | ||
if(_session != null) | ||
await _session.CloseAsync(); | ||
|
||
if(_connection != null) | ||
await _connection.CloseAsync(); | ||
} | ||
|
||
public SenderLink GetSenderLink() | ||
{ | ||
if (_connection == null || _connection.IsClosed) | ||
_connection = new Connection(_address); | ||
|
||
if (_session == null || _session.IsClosed) | ||
_session = new Session(_connection); | ||
|
||
return new SenderLink( | ||
_session, | ||
_linkName, | ||
new Target | ||
{ | ||
Address = _queueName, | ||
Capabilities = new[] { new Symbol("queue") } | ||
}, | ||
null); | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,79 @@ | ||
using System; | ||
using System.Collections.Generic; | ||
using System.Text; | ||
using System.Threading.Tasks; | ||
using Akka.Serialization; | ||
using Amqp; | ||
using Amqp.Framing; | ||
using Amqp.Types; | ||
|
||
namespace Akka.Streams.Amqp.V1 | ||
{ | ||
public class AddressSourceSettings<T> : IAmqpSourceSettings<T> | ||
{ | ||
private readonly string _linkName; | ||
private readonly string _queueName; | ||
private readonly Serializer _serializer; | ||
|
||
public int Credit { get; } | ||
private readonly Address _address; | ||
private Connection _connection; | ||
private Session _session; | ||
|
||
public bool ManageConnection => true; | ||
|
||
public AddressSourceSettings( | ||
Address address, | ||
string linkName, | ||
string queueName, | ||
int credit, | ||
Serializer serializer) | ||
{ | ||
_address = address; | ||
_linkName = linkName; | ||
_queueName = queueName; | ||
Credit = credit; | ||
_serializer = serializer; | ||
} | ||
|
||
public T Convert(Message message) | ||
{ | ||
var bString = message.GetBody<byte[]>(); | ||
return _serializer.FromBinary<T>(bString); | ||
} | ||
|
||
public void CloseConnection() | ||
{ | ||
_session?.Close(); | ||
_connection?.Close(); | ||
} | ||
|
||
public async Task CloseConnectionAsync() | ||
{ | ||
if (_session != null) | ||
await _session.CloseAsync(); | ||
|
||
if (_connection != null) | ||
await _connection.CloseAsync(); | ||
} | ||
|
||
public ReceiverLink GetReceiverLink() | ||
{ | ||
if(_connection == null || _connection.IsClosed) | ||
_connection = new Connection(_address); | ||
|
||
if (_session == null || _session.IsClosed) | ||
_session = new Session(_connection); | ||
|
||
return new ReceiverLink( | ||
_session, | ||
_linkName, | ||
new Source | ||
{ | ||
Address = _queueName, | ||
Capabilities = new[] { new Symbol("queue") } | ||
}, | ||
null); | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,11 +1,25 @@ | ||
using Akka.Streams.Stage; | ||
using System; | ||
using System.Collections.Generic; | ||
using Amqp; | ||
using System.Threading.Tasks; | ||
using Amqp.Framing; | ||
|
||
namespace Akka.Streams.Amqp.V1 | ||
{ | ||
public sealed class AmqpSinkStage<T> : GraphStageWithMaterializedValue<SinkShape<T>, Task> | ||
{ | ||
private static readonly Dictionary<int, TimeSpan> RetryInterval = | ||
new Dictionary<int, TimeSpan>() | ||
{ | ||
{ 6, TimeSpan.FromMilliseconds(100) }, | ||
{ 5, TimeSpan.FromMilliseconds(500) }, | ||
{ 4, TimeSpan.FromMilliseconds(1000) }, | ||
{ 3, TimeSpan.FromMilliseconds(2000) }, | ||
{ 2, TimeSpan.FromMilliseconds(4000) }, | ||
{ 1, TimeSpan.FromMilliseconds(8000) }, | ||
}; | ||
|
||
public Inlet<T> In { get; } | ||
public override SinkShape<T> Shape { get; } | ||
public IAmqpSinkSettings<T> AmqpSourceSettings { get; } | ||
|
@@ -28,13 +42,15 @@ private class AmqpSinkStageLogic : GraphStageLogic | |
{ | ||
private readonly AmqpSinkStage<T> _stage; | ||
private readonly TaskCompletionSource<Done> _promise; | ||
private readonly SenderLink _sender; | ||
private readonly Action<(IAmqpObject, Error)> _disconnectedCallback; | ||
|
||
private SenderLink _sender; | ||
|
||
public AmqpSinkStageLogic(AmqpSinkStage<T> amqpSinkStage, TaskCompletionSource<Done> promise, SinkShape<T> shape) : base(shape) | ||
{ | ||
_stage = amqpSinkStage; | ||
_promise = promise; | ||
_sender = amqpSinkStage.AmqpSourceSettings.GetSenderLink(); | ||
_disconnectedCallback = GetAsyncCallback<(IAmqpObject, Error)>(HandleDisconnection); | ||
|
||
SetHandler( | ||
inlet: _stage.In, | ||
|
@@ -49,15 +65,52 @@ public AmqpSinkStageLogic(AmqpSinkStage<T> amqpSinkStage, TaskCompletionSource<D | |
); | ||
} | ||
|
||
private async Task Connect() | ||
{ | ||
var retry = 7; | ||
var exceptions = new List<Exception>(); | ||
while (true) | ||
{ | ||
try | ||
{ | ||
_sender = _stage.AmqpSourceSettings.GetSenderLink(); | ||
_sender.AddClosedCallback((sender, error) => _disconnectedCallback((sender, error))); | ||
Log.Info("Connected to AMQP.V1 server."); | ||
return; | ||
} | ||
catch (Exception e) | ||
{ | ||
if (!_stage.AmqpSourceSettings.ManageConnection) | ||
{ | ||
throw new ConnectionException( | ||
"Failed to connect to AMQP.V1 server. Could not retry connection because SinkSettings does not manage the Connection object.", e); | ||
} | ||
|
||
retry--; | ||
if (retry == 0) | ||
throw new AggregateException("Failed to connect to AMQP.V1 server.", exceptions); | ||
|
||
exceptions.Add(e); | ||
Log.Error($"[{retry}] more retries to connect to AMQP.V1 server."); | ||
await Task.Delay(RetryInterval[retry]); | ||
} | ||
} | ||
} | ||
|
||
private void HandleDisconnection((IAmqpObject sender, Error error) args) | ||
=> FailStage(new DisconnectedException(args.sender, args.error)); | ||
|
||
public override void PreStart() | ||
{ | ||
base.PreStart(); | ||
|
||
Connect().Wait(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Same comment as below about blocking on a really long-running call |
||
Pull(_stage.In); | ||
} | ||
|
||
public override void PostStop() | ||
{ | ||
_sender.Close(); | ||
_sender?.Close(); | ||
base.PostStop(); | ||
} | ||
} | ||
|
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM