Return amqp channel with RabbitMQ client

go.mod

@@ -19,20 +19,20 @@ require (
 	github.com/gocql/gocql v1.6.0
 	github.com/grafadruid/go-druid v0.0.6
 	github.com/lib/pq v1.10.7
-	github.com/michaelklishin/rabbit-hole/v2 v2.16.0
+	github.com/michaelklishin/rabbit-hole/v3 v3.1.0
 	github.com/microsoft/go-mssqldb v1.6.0
 	github.com/opensearch-project/opensearch-go v1.1.0
 	github.com/opensearch-project/opensearch-go/v2 v2.3.0
 	github.com/pkg/errors v0.9.1
-	github.com/rabbitmq/amqp091-go v1.9.0
+	github.com/rabbitmq/amqp091-go v1.10.0
 	github.com/redis/go-redis/v9 v9.5.1
 	go.mongodb.org/mongo-driver v1.14.0
 	k8s.io/api v0.30.3
 	k8s.io/apimachinery v0.30.3
 	k8s.io/klog/v2 v2.130.1
-	kmodules.xyz/client-go v0.30.38
+	kmodules.xyz/client-go v0.30.41
 	kmodules.xyz/custom-resources v0.30.0
-	kubedb.dev/apimachinery v0.49.0
+	kubedb.dev/apimachinery v0.49.1-0.20241217045444-52bb83407e60
 	sigs.k8s.io/controller-runtime v0.18.4
 	xorm.io/xorm v1.3.6
 )
@@ -43,7 +43,7 @@ require (
 	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
-	github.com/cert-manager/cert-manager v1.15.2 // indirect
+	github.com/cert-manager/cert-manager v1.15.4 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/coreos/go-semver v0.3.1 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
@@ -120,14 +120,14 @@ require (
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.opentelemetry.io/otel v1.28.0 // indirect
 	go.opentelemetry.io/otel/trace v1.28.0 // indirect
-	golang.org/x/crypto v0.27.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
-	golang.org/x/net v0.29.0 // indirect
+	golang.org/x/net v0.30.0 // indirect
 	golang.org/x/oauth2 v0.22.0 // indirect
-	golang.org/x/sync v0.8.0 // indirect
-	golang.org/x/sys v0.25.0 // indirect
-	golang.org/x/term v0.24.0 // indirect
-	golang.org/x/text v0.18.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/term v0.27.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	gomodules.xyz/mergo v0.3.13 // indirect
@@ -146,6 +146,7 @@ require (
 	kmodules.xyz/monitoring-agent-api v0.30.2 // indirect
 	kmodules.xyz/offshoot-api v0.30.1 // indirect
 	kubeops.dev/petset v0.0.7 // indirect
+	kubeops.dev/sidekick v0.0.10-0.20241122131943-163e27e5ef71 // indirect
 	modernc.org/memory v1.5.0 // indirect
 	modernc.org/token v1.1.0 // indirect
 	sigs.k8s.io/gateway-api v1.1.0 // indirect

rabbitmq/client.go

@@ -17,13 +17,14 @@ limitations under the License.
 package rabbitmq
 
 import (
-	rmqhttp "github.com/michaelklishin/rabbit-hole/v2"
+	rmqhttp "github.com/michaelklishin/rabbit-hole/v3"
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 type Client struct {
 	AMQPClient
 	HTTPClient
+	Channel
 }
 
 type AMQPClient struct {
@@ -37,3 +38,33 @@ type HTTPClient struct {
 type Channel struct {
 	*amqp.Channel
 }
+
+type ConnectionQueue struct {
+	conn map[string]*Client
+}
+
+func NewConnectionQueue() *ConnectionQueue {
+	return &ConnectionQueue{
+		conn: make(map[string]*Client),
+	}
+}
+
+func (c *ConnectionQueue) GetAMQPConnection(key string) *AMQPClient {
+	return &c.conn[key].AMQPClient
+}
+
+func (c *ConnectionQueue) GetHTTPConnection(key string) *HTTPClient {
+	return &c.conn[key].HTTPClient
+}
+
+func (c *ConnectionQueue) GetAMQPChannel(key string) *Channel {
+	return &c.conn[key].Channel
+}
+
+func (c *ConnectionQueue) GetClientWithKey(key string) *Client {
+	return c.conn[key]
+}
+
+func (c *ConnectionQueue) SetClientWithKey(key string, client *Client) {
+	c.conn[key] = client
+}

rabbitmq/http_client.go

@@ -19,7 +19,7 @@ package rabbitmq
 import (
 	"fmt"
 
-	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+	rmqhttp "github.com/michaelklishin/rabbit-hole/v3"
 	"k8s.io/klog/v2"
 )
 
@@ -45,7 +45,7 @@ func (c *HTTPClient) IsAllNodesRunningInCluster(replicas int) (bool, error) {
 	return true, nil
 }
 
-func (c *HTTPClient) GetQueues() ([]rabbithole.QueueInfo, error) {
+func (c *HTTPClient) GetQueues() ([]rmqhttp.QueueInfo, error) {
 	queues, err := c.Client.ListQueues()
 	if err != nil {
 		klog.Error(err, "Failed to get queue lists")
@@ -54,13 +54,13 @@ func (c *HTTPClient) GetQueues() ([]rabbithole.QueueInfo, error) {
 	return queues, nil
 }
 
-func (c *HTTPClient) GetClassicQueues() ([]rabbithole.QueueInfo, error) {
+func (c *HTTPClient) GetClassicQueues() ([]rmqhttp.QueueInfo, error) {
 	queues, err := c.GetQueues()
 	if err != nil {
 		klog.Error(err, "Failed to get queue lists")
 		return nil, err
 	}
-	classicQueues := []rabbithole.QueueInfo{}
+	classicQueues := []rmqhttp.QueueInfo{}
 
 	for _, q := range queues {
 		if q.Type == rabbitmqQueueTypeClassic {
@@ -71,7 +71,7 @@ func (c *HTTPClient) GetClassicQueues() ([]rabbithole.QueueInfo, error) {
 	return classicQueues, nil
 }
 
-func (c *HTTPClient) HasNodeAnyClassicQueue(queues []rabbithole.QueueInfo, node string) bool {
+func (c *HTTPClient) HasNodeAnyClassicQueue(queues []rmqhttp.QueueInfo, node string) bool {
 	for _, q := range queues {
 		if q.Type == rabbitmqQueueTypeClassic && q.Node == node {
 			return true
@@ -80,13 +80,13 @@ func (c *HTTPClient) HasNodeAnyClassicQueue(queues []rabbithole.QueueInfo, node
 	return false
 }
 
-func (c *HTTPClient) GetQuorumQueues() ([]rabbithole.QueueInfo, error) {
+func (c *HTTPClient) GetQuorumQueues() ([]rmqhttp.QueueInfo, error) {
 	queues, err := c.GetQueues()
 	if err != nil {
 		klog.Error(err, "Failed to get queue lists")
 		return nil, err
 	}
-	quorumQueues := []rabbithole.QueueInfo{}
+	quorumQueues := []rmqhttp.QueueInfo{}
 
 	for _, q := range queues {
 		if q.Type == rabbitmqQueueTypeQuorum {
@@ -97,7 +97,7 @@ func (c *HTTPClient) GetQuorumQueues() ([]rabbithole.QueueInfo, error) {
 	return quorumQueues, nil
 }
 
-func (c *HTTPClient) IsNodePrimaryReplica(queues []rabbithole.QueueInfo, node string) bool {
+func (c *HTTPClient) IsNodePrimaryReplica(queues []rmqhttp.QueueInfo, node string) bool {
 	for _, q := range queues {
 		if q.Type == rabbitmqQueueTypeQuorum && q.Leader == node {
 			return true

rabbitmq/kubedb_client_builder.go

@@ -27,7 +27,7 @@ import (
 	"strings"
 	"time"
 
-	rmqhttp "github.com/michaelklishin/rabbit-hole/v2"
+	rmqhttp "github.com/michaelklishin/rabbit-hole/v3"
 	amqp "github.com/rabbitmq/amqp091-go"
 	core "k8s.io/api/core/v1"
 	kerr "k8s.io/apimachinery/pkg/api/errors"
@@ -47,6 +47,7 @@ type KubeDBClientBuilder struct {
 	httpURL           string
 	podName           string
 	vhost             string
+	connName          string
 	enableHTTPClient  bool
 	disableAMQPClient bool
 }
@@ -92,6 +93,11 @@ func (o *KubeDBClientBuilder) WithVHost(vhost string) *KubeDBClientBuilder {
 	return o
 }
 
+func (o *KubeDBClientBuilder) WithConnectionName(connName string) *KubeDBClientBuilder {
+	o.connName = connName
+	return o
+}
+
 func (o *KubeDBClientBuilder) WithContext(ctx context.Context) *KubeDBClientBuilder {
 	o.ctx = ctx
 	return o
@@ -215,16 +221,31 @@ func (o *KubeDBClientBuilder) GetRabbitMQClient() (*Client, error) {
 			o.amqpURL = o.GetAMQPconnURL(username, password, o.vhost)
 		}
 
+		extraConfigProperties := amqp.NewConnectionProperties()
+		if o.connName != "" {
+			extraConfigProperties.SetClientConnectionName(o.connName)
+		}
+
 		rabbitConnection, err := amqp.DialConfig(o.amqpURL, amqp.Config{
-			Vhost:  o.vhost,
-			Locale: "en_US",
+			Vhost:      o.vhost,
+			Locale:     "en_US",
+			Properties: extraConfigProperties,
 		})
 		if err != nil {
 			klog.Error(err, "Failed to connect to rabbitmq")
 			return nil, err
 		}
 		klog.Info("Successfully created AMQP client for RabbitMQ")
+
+		msgChannel, err := rabbitConnection.Channel()
+		if err != nil {
+			klog.Error(err, "Failed to create AMQP channel")
+			return nil, err
+		}
+		klog.Info("Successfully created AMQP channel for RabbitMQ")
+
 		rmqClient.AMQPClient = AMQPClient{rabbitConnection}
+		rmqClient.Channel = Channel{msgChannel}
 	}
 
 	return rmqClient, nil