Feature request: connection automatic reconnect

[ChugR]

Issue #32 discussed this. It is a common pattern that many amqpnetlite users will see and it would be a great feature to consider. A model for this feature is Qpid C++ Messaging which describes reconnect behavior and option settings.

[ChugR]

A second feature request built on top of automatic reconnect is failover. A user could send a list of Address objects to a Connection constructor instead of a single Address.

The library would try to connect to each address in turn.

[xinchen10]

When connection reconnect is enabled, does it just re-establish the transport? How about sessions/links?

I think it makes sense to build these features in a layer on top of the protocol stack. For example, the Service Bus .NET SDK has MessagingFactory/Sender/Receiver objects. They auto-create the underlying connection/session/link when needed. The Proton Messenger might do something similar.

[mcmellawatt]

We would very much like this feature as well. Currently we have built our own reconnection logic which recreates the session/link/connection objects when certain AmqpExceptions are caught in the client. It works, though I am not sure if it would make more sense to handle reconnection at a lower level.

If our approach is reasonable, from my understanding there would be no need to call Close() on the existing link/session/connection objects before creating new ones, because closing the ContainerHost on the server or terminating the server process actually triggers closing/aborting logic in the client. Is this a reasonable assumption?

[xinchen10]

Thanks for the feedback. We will consider enabling this feature in a future release (cannot promise when it will be ready though). Application would still need to recreate session/link in cases of communication errors.

[networkfusion]

@mcmellawatt would you be prepared to share your reconnection wrapper/logic? I am having trouble with keeping connections open over long periods of time....

[caserzer]

@mcmellawatt I have the same question , would you share the retry logic ?
@networkfusion can you share the exception inform?

[duizendnegen]

Similarly so, my listener client just stops working at some point - I seem not even to receive a Close event so I can't decide to reconnect or not.

[duizendnegen]

I created my own reconnection logic, also for the receiver objects (which never catches exceptions, since it merely stops receiving messages) and added this to a demo project. See https://github.com/duizendnegen/net-docker-web-worker-role/blob/master/Jobs/QueueListener.cs and https://github.com/duizendnegen/net-docker-web-worker-role/blob/master/Jobs/QueueConnection.cs.
I'll make a write-up of this in a blog post shortly.

[duizendnegen]

Completed the write-up, in case anyone stumbles on the same problem: https://medium.com/@duizendnegen/running-worker-roles-with-docker-in-net-core-543b8d1c4ae7

[xinchen10]

I added an article that describes common reconnect approaches. There is also a test project that demonstrates the usage. I am able to run the test against a Service Bus queue for weeks and transfer tens of millions messages between the senders and the receivers.

[duizendnegen]

Thanks for the write-up @xinchen10, seems like a holistic approach to solving this problem.
Reading through the code and also the documentation I couldn't quite figure out how the Consumer re-connects after the 10 minute auto-disconnect after inactivity from Service Queue kicks in. As far as I know, the Closed event doesn't fire in this case, nor is the IsClosed property checked (because no new Consumer is constructed, the only thing that it notices is that no more messages are arriving.) Could you elaborate how this case is handled?

[xinchen10]

@duizendnegen Consumer supports three modes for receiving messages: sync, async and callback. The base class Role.SyncTest/AsyncTest has a while loop that drives the test execution by doing EnsureConnection, EnsureLink and Execute repeatedly. The sync and async modes are straightforward. The callback mode is a bit confusing because it does not need a while loop to drive execution, but I used the loop in the base class to run auto-reconnect. The interesting class is Consumer.MyCallbackTest. The CreateLink method subscribes to the Closed event of all Connection, Session and ReceiverLink objects and the Execute method blocks the execution until the manual reset event is signaled, which is done when any of the AMQP objects is closed. In real code this logic may be just inside OnChannelClosed event handler. The while loop is single threaded so I do not need to worry about concurrency issue. If the reconnect code could be invoked concurrently, the application must have some synchronization to ensure an AMQP object is recreated only once.

You are right that the Service Bus queue closes idle receivers after 10 minutes but if a receiver has outstanding link credits it is considered active and will not be closed by the idle timer (but it could be closed for other reasons).

To test the Closed event of the receiver, I tried the following code:

static void Main(string[] args)
{
    var a = new Address("amqps://<key>:<value>@<myns>.servicebus.windows.net");
    var c = Connection.Factory.CreateAsync(a).Result;
    var s = new Session(c);
    var r = new ReceiverLink(s, "receiver1", "q1");
    r.Closed += R_Closed;

    Thread.Sleep(-1);
    c.Close();
}

private static void R_Closed(AmqpObject sender, Error error)
{
    Console.WriteLine(sender + " closed");
}

Note that I did not call Start or Receive so the receiver link does not send any credits to the service. After 10 minutes it is closed and the event handler is fired. Did you subscribe to the Closed event of Connection and Session in your code?

[duizendnegen]

@xinchen10 you are right, I adapted the example to also test still being able to receive new messages (i.e. with outstanding credits) and that works perfectly - I'm not sure why originally I hadn't been receiving the Closed event, but that seems to be entirely on my side.

Just for completeness, the Program.cs I ended up using was

private static Address a;

static void Main(string[] args)
{
    a = new Address("queue-experiment.servicebus.windows.net",
        5671,
        "SendListen",
        "W/w7Kf2g33nf0nDbe9gvLN68lCkGb8ZhQPk2WPB4fzI=");

    Thread publisher = new Thread(PublisherThread);
    Thread consumer = new Thread(ConsumerThread);

    publisher.Start();
    consumer.Start();

    publisher.Join();
    consumer.Join();
}

private static void PublisherThread()
{
    {
        var c = Connection.Factory.CreateAsync(a).Result;
        var s = new Session(c);
        var p = new SenderLink(s, "sender1", "experiment");
        p.Send(new Message("Send 1"));
        Console.WriteLine("Send 1 sent");
    }

    // sleep 60 minutes
    Thread.Sleep(1000 * 60 * 60);

    {
        var c = Connection.Factory.CreateAsync(a).Result;
        var s = new Session(c);
        var p = new SenderLink(s, "sender2", "experiment");
        p.Send(new Message("Send 2"));
        Console.WriteLine("Send 2 sent");
    }
}

private static void ConsumerThread()
{
    var c = Connection.Factory.CreateAsync(a).Result;
    var s = new Session(c);
    var r = new ReceiverLink(s, "receiver1", "experiment");
    r.Closed += R_Closed;

    r.Start(10, R_OnMessage);

    Thread.Sleep(-1);
    c.Close();
}

private static void R_OnMessage(ReceiverLink receiver, Message message)
{
    Console.WriteLine(message.GetBody<string>() + " received");
    receiver.Accept(message);
}

private static void R_Closed(AmqpObject sender, Error error)
{
    Console.WriteLine(sender + " closed");
}

where I do not receive the closed event and still receive a new message after 1 hour, as desired.

[brvaland]

@xinchen10 @duizendnegen - I am a bit confused reference to outstanding credit specified on receiver.start as how it works. Does service bus closes the connection once the outstanding credit has reached?

So lets say you have specified a credit of 10, so you will be receiving 10 messages without any time restriction but what happens after you have received 10 messages will it reconnect automatically with new credit of 10.

Please clarify?