Skip to content

Commit

Permalink
Back-port rabbitmq#1616 to 6.x
Browse files Browse the repository at this point in the history
Fixes rabbitmq#1617

Update RecordedConsumer.cs
Update RecordedExchange.cs
Update RecordedQueue.cs
Update projects/RabbitMQ.Client/client/impl/RecordedBinding.cs
Update projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs
Update projects/RabbitMQ.Client/client/impl/RecordedExchange.cs
Update projects/RabbitMQ.Client/client/impl/RecordedQueue.cs
Update projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs
Update projects/RabbitMQ.Client/client/impl/RecordedBinding.cs
Update projects/RabbitMQ.Client/client/impl/RecordedExchange.cs
Update projects/RabbitMQ.Client/client/impl/RecordedQueue.cs
Update `TestRecoveringConsumerHandlerOnConnection_EventArgumentsArePassedDown`
Update projects/Unit/TestConnectionRecovery.cs

Port test from `main`
  • Loading branch information
Saibamen authored and lukebakken committed Jun 28, 2024
1 parent 8f32ba3 commit 15a87c0
Show file tree
Hide file tree
Showing 5 changed files with 94 additions and 9 deletions.
10 changes: 9 additions & 1 deletion projects/RabbitMQ.Client/client/impl/RecordedBinding.cs
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,15 @@ public override string ToString()

public RecordedBinding WithArguments(IDictionary<string, object> value)
{
Arguments = value;
if (value is null)
{
Arguments = null;
}
else
{
Arguments = new Dictionary<string, object>(value);
}

return this;
}

Expand Down
10 changes: 9 additions & 1 deletion projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,15 @@ public string Recover(IModel channelToUse)

public RecordedConsumer WithArguments(IDictionary<string, object> value)
{
Arguments = value;
if (value is null)
{
Arguments = null;
}
else
{
Arguments = new Dictionary<string, object>(value);
}

return this;
}

Expand Down
10 changes: 9 additions & 1 deletion projects/RabbitMQ.Client/client/impl/RecordedExchange.cs
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,15 @@ public override string ToString()

public RecordedExchange WithArguments(IDictionary<string, object> value)
{
Arguments = value;
if (value is null)
{
Arguments = null;
}
else
{
Arguments = new Dictionary<string, object>(value);
}

return this;
}

Expand Down
10 changes: 9 additions & 1 deletion projects/RabbitMQ.Client/client/impl/RecordedQueue.cs
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,15 @@ protected string NameToUseForRecovery

public RecordedQueue WithArguments(IDictionary<string, object> value)
{
Arguments = value;
if (value is null)
{
Arguments = null;
}
else
{
Arguments = new Dictionary<string, object>(value);
}

return this;
}

Expand Down
63 changes: 58 additions & 5 deletions projects/Unit/TestConnectionRecovery.cs
Original file line number Diff line number Diff line change
Expand Up @@ -744,10 +744,17 @@ public void TestRecoveringConsumerHandlerOnConnection()
[Test]
public void TestRecoveringConsumerHandlerOnConnection_EventArgumentsArePassedDown()
{
var myArgs = new Dictionary<string, object> { { "first-argument", "some-value" } };
const string key = "first-argument";
const string value = "some-value";

IDictionary<string, object> arguments = new Dictionary<string, object>
{
{ key, value }
};

string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
var cons = new EventingBasicConsumer(Model);
string expectedCTag = Model.BasicConsume(cons, q, arguments: myArgs);
string expectedCTag = Model.BasicConsume(cons, q, arguments: arguments);

bool ctagMatches = false;
bool consumerArgumentMatches = false;
Expand All @@ -757,14 +764,15 @@ public void TestRecoveringConsumerHandlerOnConnection_EventArgumentsArePassedDow
// passed to a CallbackExceptionHandler, instead of failing the test. Instead, we have to do this trick
// and assert in the test function.
ctagMatches = args.ConsumerTag == expectedCTag;
consumerArgumentMatches = (string)args.ConsumerArguments["first-argument"] == "some-value";
args.ConsumerArguments["first-argument"] = "event-handler-set-this-value";
consumerArgumentMatches = (string)args.ConsumerArguments[key] == value;
};

CloseAndWaitForRecovery();
Assert.That(ctagMatches, Is.True, "expected consumer tag to match");
Assert.That(consumerArgumentMatches, Is.True, "expected consumer arguments to match");
Assert.That(myArgs, Does.ContainKey("first-argument").WithValue("event-handler-set-this-value"));
Assert.That(arguments.ContainsKey(key), Is.True);
string actualVal = (string)arguments[key];
Assert.That(actualVal, Is.EqualTo(value));
}

[Test]
Expand Down Expand Up @@ -1687,6 +1695,51 @@ void MessageReceived(object sender, BasicDeliverEventArgs e)
}
}

[Test]
public void TestQueueRecoveryWithDlxArgument_RabbitMQUsers_hk5pJ4cKF0c()
{
string tdiWaitExchangeName = GenerateExchangeName();
string tdiRetryExchangeName = GenerateExchangeName();
string testRetryQueueName = GenerateQueueName();
string testQueueName = GenerateQueueName();

Model.ExchangeDeclare(exchange: tdiWaitExchangeName,
type: ExchangeType.Topic, durable: true, autoDelete: false, arguments: null);
Model.ExchangeDeclare(exchange: tdiRetryExchangeName,
type: ExchangeType.Topic, durable: true, autoDelete: false, arguments: null);

var arguments = new Dictionary<string, object>
{
{ "x-dead-letter-exchange", "tdi.retry.exchange" },
{ "x-dead-letter-routing-key", "QueueTest" }
};

Model.QueueDeclare(testRetryQueueName, durable: false, exclusive: false, autoDelete: false, arguments);

arguments["x-dead-letter-exchange"] = "tdi.wait.exchange";
arguments["x-dead-letter-routing-key"] = "QueueTest";

Model.QueueDeclare(testQueueName, durable: false, exclusive: false, autoDelete: false, arguments);

arguments.Remove("x-dead-letter-exchange");
arguments.Remove("x-dead-letter-routing-key");

Model.QueueBind(testRetryQueueName, tdiWaitExchangeName, testQueueName);

Model.QueueBind(testQueueName, tdiRetryExchangeName, testQueueName);

var consumerAsync = new EventingBasicConsumer(Model);
Model.BasicConsume(queue: testQueueName, autoAck: false, consumer: consumerAsync);

CloseAndWaitForRecovery();

QueueDeclareOk q0 = Model.QueueDeclarePassive(testRetryQueueName);
Assert.AreEqual(testRetryQueueName, q0.QueueName);

QueueDeclareOk q1 = Model.QueueDeclarePassive(testQueueName);
Assert.AreEqual(testQueueName, q1.QueueName);
}

internal bool SendAndConsumeMessage(string queue, string exchange, string routingKey)
{
using (var ch = Conn.CreateModel())
Expand Down

0 comments on commit 15a87c0

Please sign in to comment.