Skip to content
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 event for recovering consumer #1304

Merged
merged 2 commits into from
Feb 23, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -45,5 +45,6 @@ public interface IAutorecoveringConnection : IConnection

event EventHandler<ConsumerTagChangedAfterRecoveryEventArgs> ConsumerTagChangeAfterRecovery;
event EventHandler<QueueNameChangedAfterRecoveryEventArgs> QueueNameChangeAfterRecovery;
event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
// This source code is dual-licensed under the Apache License, version
// 2.0, and the Mozilla Public License, version 2.0.
//
// The APL v2.0:
//
//---------------------------------------------------------------------------
// Copyright (c) 2007-2023 VMware, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//---------------------------------------------------------------------------
//
// The MPL v2.0:
//
//---------------------------------------------------------------------------
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.
//
// Copyright (c) 2007-2023 VMware, Inc. All rights reserved.
//---------------------------------------------------------------------------

using System;
using System.Collections.Generic;

namespace RabbitMQ.Client.Events
{
/// <summary>
/// Event related to consumer recovery, during automatic recovery.
/// </summary>
public class RecoveringConsumerEventArgs : EventArgs
{
/// <summary>
/// Constructs an event containing the consumer arguments and consumer
/// tag of the consumer this event relates to.
/// </summary>
/// <param name="consumerArguments">Consumer arguments of the consumer for this event</param>
/// <param name="consumerTag">Consumer tag of the consumer for this event</param>
public RecoveringConsumerEventArgs(IDictionary<string, object> consumerArguments, string consumerTag)
{
ConsumerArguments = consumerArguments;
ConsumerTag = consumerTag;
}

/// <summary>
/// Access the consumer arguments of the consumer this event relates to.
/// </summary>
public IDictionary<string, object> ConsumerArguments { get; }

/// <summary>
/// Access the consumer tag of the consumer this event relates to.
/// </summary>
public string ConsumerTag { get; }
}
}
15 changes: 15 additions & 0 deletions projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,7 @@ public event EventHandler<EventArgs> ConnectionUnblocked
}
}

public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer;
public event EventHandler<ConsumerTagChangedAfterRecoveryEventArgs> ConsumerTagChangeAfterRecovery;
public event EventHandler<QueueNameChangedAfterRecoveryEventArgs> QueueNameChangeAfterRecovery;

Expand Down Expand Up @@ -1099,6 +1100,20 @@ internal void RecoverConsumers(AutorecoveringModel modelToRecover, IModel channe
string tag = pair.Key;
try
{
foreach (EventHandler<RecoveringConsumerEventArgs> eh in RecoveringConsumer?.GetInvocationList() ?? Array.Empty<Delegate>())
{
try
{
var eventArgs = new RecoveringConsumerEventArgs(cons.Arguments, cons.ConsumerTag);
eh(this, eventArgs);
}
catch (Exception e)
{
var args = new CallbackExceptionEventArgs(e) { Detail = { ["context"] = "OnBeforeRecoveringConsumer" } };
_delegate.OnCallbackException(args);
}
}

string newTag = cons.Recover(channelToUse);
lock (_recordedConsumers)
{
Expand Down
7 changes: 7 additions & 0 deletions projects/Unit/APIApproval.Approve.verified.txt
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,7 @@ namespace RabbitMQ.Client
event System.EventHandler<RabbitMQ.Client.Events.ConnectionRecoveryErrorEventArgs> ConnectionRecoveryError;
event System.EventHandler<RabbitMQ.Client.Events.ConsumerTagChangedAfterRecoveryEventArgs> ConsumerTagChangeAfterRecovery;
event System.EventHandler<RabbitMQ.Client.Events.QueueNameChangedAfterRecoveryEventArgs> QueueNameChangeAfterRecovery;
event System.EventHandler<RabbitMQ.Client.Events.RecoveringConsumerEventArgs> RecoveringConsumer;
event System.EventHandler<System.EventArgs> RecoverySucceeded;
}
public interface IBasicConsumer
Expand Down Expand Up @@ -689,6 +690,12 @@ namespace RabbitMQ.Client.Events
public string NameAfter { get; }
public string NameBefore { get; }
}
public class RecoveringConsumerEventArgs : System.EventArgs
{
public RecoveringConsumerEventArgs(System.Collections.Generic.IDictionary<string, object> consumerArguments, string consumerTag) { }
public System.Collections.Generic.IDictionary<string, object> ConsumerArguments { get; }
public string ConsumerTag { get; }
}
public class RecoveryExceptionEventArgs : RabbitMQ.Client.Events.BaseExceptionEventArgs
{
public RecoveryExceptionEventArgs(System.Exception e) { }
Expand Down
47 changes: 47 additions & 0 deletions projects/Unit/TestConnectionRecovery.cs
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
//---------------------------------------------------------------------------

using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;

Expand Down Expand Up @@ -688,6 +689,52 @@ public void TestRecoveryEventHandlersOnConnection()
Assert.IsTrue(counter >= 3);
}

[Test]
public void TestRecoveringConsumerHandlerOnConnection()
{
string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
var cons = new EventingBasicConsumer(Model);
Model.BasicConsume(q, true, cons);

int counter = 0;
((AutorecoveringConnection)Conn).RecoveringConsumer += (sender, args) => Interlocked.Increment(ref counter);

CloseAndWaitForRecovery();
Assert.IsTrue(Conn.IsOpen, "expected connection to be open");
Assert.AreEqual(1, counter);

CloseAndWaitForRecovery();
CloseAndWaitForRecovery();
Assert.IsTrue(Conn.IsOpen, "expected connection to be open");
Assert.AreEqual(3, counter);
}

[Test]
public void TestRecoveringConsumerHandlerOnConnection_EventArgumentsArePassedDown()
{
var myArgs = new Dictionary<string, object> { { "first-argument", "some-value" } };
string q = Model.QueueDeclare(GenerateQueueName(), false, false, false, null).QueueName;
var cons = new EventingBasicConsumer(Model);
string expectedCTag = Model.BasicConsume(cons, q, arguments: myArgs);

bool ctagMatches = false;
bool consumerArgumentMatches = false;
((AutorecoveringConnection)Conn).RecoveringConsumer += (sender, args) =>
{
// We cannot assert here because NUnit throws when an assertion fails. This exception is caught and
// 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";
};

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"));
}

[Test]
public void TestRecoveryEventHandlersOnModel()
{
Expand Down