pubsub: automatically recovering RabbitMQ connections

[kmlebedev]

Describe the solution you'd like

Now the connection is established once during the initialization of the defaultDialer and there is no way to restore it without restarting the application

Describe alternatives you've considered

Automatically recovering RabbitMQ connections

Additional context

https://medium.com/@dhanushgopinath/automatically-recovering-rabbitmq-connections-in-go-applications-7795a605ca59

go-cloud/pubsub/rabbitpubsub/rabbit.go

Line 56 in f28b854

conn, err := amqp.Dial(serverURL)

[kmlebedev]

I tried to reconnect, but it is not clear how to reset the saved error
"Non-retryable error from ReceiveBatch -> permanent error."
https://github.com/google/go-cloud/blob/master/pubsub/pubsub.go#L560

[vangent]

Just restart the application on any permanent error and it will reconnect.

[kmlebedev]

Just restart the application on any permanent error and it will reconnect.

I can afford to reload the message receiving application
But an application that sends messages is not acceptable

Need to invent something
since the reconnection cannot be obtained on the application side

[vangent]

We do already recreate the amqp channel as needed using NotifyClose (search for closec in rabbit.go).

You can create your own URLOpener if you don't want to use the default one:

go-cloud/pubsub/rabbitpubsub/rabbit.go

Line 93 in f28b854

type URLOpener struct {

Or, use the explicit constructor where you can provide the connection yourself:

go-cloud/pubsub/rabbitpubsub/rabbit.go

Line 154 in f28b854

func OpenTopic(conn *amqp.Connection, name string, opts *TopicOptions) *pubsub.Topic {

Does that help?

[vangent]

I supposed defaultConn could save the URL it used, and if the connection is IsClosed reconnect?

go-cloud/pubsub/rabbitpubsub/rabbit.go

Line 62 in f28b854

})

https://pkg.go.dev/github.com/streadway/amqp#Connection.IsClosed

[kmlebedev]

I supposed defaultConn could save the URL it used, and if the connection is IsClosed reconnect?

go-cloud/pubsub/rabbitpubsub/rabbit.go

Line 62 in f28b854

})

https://pkg.go.dev/github.com/streadway/amqp#Connection.IsClosed

I had no problems in this part.
Difficulty correctly throwing a new connection into Topic structure

go-cloud/pubsub/rabbitpubsub/rabbit.go

Line 121 in f28b854

type topic struct {

after initialization

[vangent]

Just create a new Topic using OpenTopic (passing a new Connection) or <your URLOpener>.OpenTopicURL (after updating the Connection field on your URLOpener).

It won't work if you are using the default URLOpener, as it only creates the connection once.

[kmlebedev]

Just create a new Topic using OpenTopic (passing a new Connection) or <your URLOpener>.OpenTopicURL (after updating the Connection field on your URLOpener).

It won't work if you are using the default URLOpener, as it only creates the connection once.

Thanks, I implemented everything on the application side, you need to do what kind of PR ?