chore: move to a maintained repo of amqp

Author: 42atomys

go.mod

@@ -12,9 +12,9 @@ require (
 	github.com/knadh/koanf/providers/file v1.1.2
 	github.com/knadh/koanf/v2 v2.1.2
 	github.com/lib/pq v1.10.9
+	github.com/rabbitmq/amqp091-go v1.10.0
 	github.com/rs/zerolog v1.33.0
 	github.com/spf13/pflag v1.0.5
-	github.com/streadway/amqp v1.1.0
 	github.com/stretchr/testify v1.9.0
 	github.com/valyala/fasthttp v1.57.0
 )

go.sum

@@ -70,6 +70,8 @@ github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5h
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
+github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
@@ -79,8 +81,6 @@ github.com/spf13/cast v1.6.0 h1:GEiTHELF+vaR5dhz3VqZfFSzZjYbgeKDpBxQVS4GYJ0=
 github.com/spf13/cast v1.6.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
-github.com/streadway/amqp v1.1.0 h1:py12iX8XSyI7aN/3dUT8DFIDJazNJsVJdxNVEpnQTZM=
-github.com/streadway/amqp v1.1.0/go.mod h1:WYSrTEYHOXHd0nwFeUXAe2G2hRnQT+deZJJf88uS9Bg=
 github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
@@ -93,6 +93,8 @@ github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVS
 github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
 github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
 github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/net v0.30.0 h1:AcW1SDZMkb8IpzCdQUaIq2sP4sZ4zw+55h6ynffypl4=
 golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

storage/rabbitmq/rabbitmq.go

@@ -7,26 +7,28 @@ import (
 	"time"
 
 	"github.com/42atomys/webhooked/internal/valuable"
+	amqp "github.com/rabbitmq/amqp091-go"
 	"github.com/rs/zerolog/log"
-	"github.com/streadway/amqp"
 )
 
 type RabbitmqStorageSpec struct {
-	DatabaseURL        valuable.Valuable `mapstructure:"databaseUrl" json:"databaseUrl"`
-	QueueName          string            `mapstructure:"queueName" json:"queueName"`
-	DefinedContentType string            `mapstructure:"contentType" json:"contentType"`
-	Durable            bool              `mapstructure:"durable" json:"durable"`
-	DeleteWhenUnused   bool              `mapstructure:"deleteWhenUnused" json:"deleteWhenUnused"`
-	Exclusive          bool              `mapstructure:"exclusive" json:"exclusive"`
-	NoWait             bool              `mapstructure:"noWait" json:"noWait"`
-	Mandatory          bool              `mapstructure:"mandatory" json:"mandatory"`
-	Immediate          bool              `mapstructure:"immediate" json:"immediate"`
-	Exchange           string            `mapstructure:"exchange" json:"exchange"`
-	MaxAttempt         int               `mapstructure:"maxAttempt" json:"maxAttempt"`
-
-	client     *amqp.Connection
-	channel    *amqp.Channel
-	routingKey amqp.Queue
+	DatabaseURL valuable.Valuable `mapstructure:"databaseUrl" json:"databaseUrl"`
+	MaxAttempt  int               `mapstructure:"maxAttempt" json:"maxAttempt"`
+	// QueueDeclare
+	QueueName        string `mapstructure:"queueName" json:"queueName"`
+	Durable          *bool  `mapstructure:"durable" json:"durable"`
+	DeleteWhenUnused bool   `mapstructure:"deleteWhenUnused" json:"deleteWhenUnused"`
+	Exclusive        bool   `mapstructure:"exclusive" json:"exclusive"`
+	NoWait           bool   `mapstructure:"noWait" json:"noWait"`
+	// Publish
+	Exchange           string `mapstructure:"exchange" json:"exchange"`
+	DefinedContentType string `mapstructure:"contentType" json:"contentType"`
+	Mandatory          bool   `mapstructure:"mandatory" json:"mandatory"`
+	Immediate          bool   `mapstructure:"immediate" json:"immediate"`
+
+	client  *amqp.Connection
+	channel *amqp.Channel
+	queue   amqp.Queue
 }
 
 func (s *RabbitmqStorageSpec) EnsureConfigurationCompleteness() error {
@@ -38,6 +40,11 @@ func (s *RabbitmqStorageSpec) EnsureConfigurationCompleteness() error {
 		s.MaxAttempt = 5
 	}
 
+	if s.Durable == nil {
+		durable := true
+		s.Durable = &durable
+	}
+
 	return nil
 }
 
@@ -61,9 +68,9 @@ func (s *RabbitmqStorageSpec) Initialize() error {
 		}
 	}()
 
-	if s.routingKey, err = s.channel.QueueDeclare(
+	if s.queue, err = s.channel.QueueDeclare(
 		s.QueueName,
-		s.Durable,
+		*s.Durable,
 		s.DeleteWhenUnused,
 		s.Exclusive,
 		s.NoWait,
@@ -77,9 +84,10 @@ func (s *RabbitmqStorageSpec) Initialize() error {
 
 func (s *RabbitmqStorageSpec) Store(ctx context.Context, value []byte) error {
 	for attempt := 0; attempt < s.MaxAttempt; attempt++ {
-		err := s.channel.Publish(
+		err := s.channel.PublishWithContext(
+			ctx,
 			s.Exchange,
-			s.routingKey.Name,
+			s.queue.Name,
 			s.Mandatory,
 			s.Immediate,
 			amqp.Publishing{