Merge branch 'main' of https://github.com/wagslane/go-rabbitmq into lw_notify

README.md

@@ -8,7 +8,13 @@ Supported by [Qvault](https://qvault.io)
 
 ## Motivation
 
-[Streadway's AMQP](https://github.com/streadway/amqp) library is currently the most robust and well-supported Go client I'm aware of. It's a fantastic option and I recommend starting there and seeing if it fulfills your needs. Their project has made an effort to stay within the scope of the AMQP protocol, as such, no reconnection logic and few ease-of-use abstractions are provided.
+[Streadway's AMQP](https://github.com/rabbitmq/amqp091-go) library is currently the most robust and well-supported Go client I'm aware of. It's a fantastic option and I recommend starting there and seeing if it fulfills your needs. Their project has made an effort to stay within the scope of the AMQP protocol, as such, no reconnection logic and few ease-of-use abstractions are provided.
+
+⚠️ **Update**  
+
+The Core Team of RabbitMQ has resumed the maintenance on [rabbitmq/amqp091-go](https://github.com/rabbitmq/amqp091-go).
+
+### Goal 
 
 The goal with `go-rabbitmq` is to still provide most all of the nitty-gritty functionality of AMQP, but to make it easier to work with via a higher-level API. Particularly:
 
@@ -128,7 +134,7 @@ Submit an issue (above in the issues tab)
 
 ## Transient Dependencies
 
-My goal is to keep dependencies limited to 1, [github.com/streadway/amqp](https://github.com/streadway/amqp).
+My goal is to keep dependencies limited to 1, [github.com/rabbitmq/amqp091-go](https://github.com/rabbitmq/amqp091-go).
 
 ## 👏 Contributing
 

channel.go

@@ -5,7 +5,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 type channelManager struct {

consume.go

@@ -4,7 +4,22 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// Action is an action that occurs after processed this delivery
+type Action int
+
+// Handler defines the handler of each Delivery and return Action
+type Handler func(d Delivery) (action Action)
+
+const (
+	// Ack default ack this msg after you have successfully processed this delivery.
+	Ack Action = iota
+	// NackDiscard the message will be dropped or delivered to a server configured dead-letter queue.
+	NackDiscard
+	// NackRequeue deliver this message to a different consumer.
+	NackRequeue
 )
 
 // Consumer allows you to create and connect to queues for data consumption.
@@ -69,7 +84,7 @@ func WithConsumerOptionsLogger(log Logger) func(options *ConsumerOptions) {
 // The provided handler is called once for each message. If the provided queue doesn't exist, it
 // will be created on the cluster
 func (consumer Consumer) StartConsuming(
-	handler func(d Delivery) bool,
+	handler Handler,
 	queue string,
 	routingKeys []string,
 	optionFuncs ...func(*ConsumeOptions),
@@ -136,7 +151,7 @@ func (consumer Consumer) StopConsuming(consumerName string, noWait bool) {
 // startGoroutinesWithRetries attempts to start consuming on a channel
 // with an exponential backoff
 func (consumer Consumer) startGoroutinesWithRetries(
-	handler func(d Delivery) bool,
+	handler Handler,
 	queue string,
 	routingKeys []string,
 	consumeOptions ConsumeOptions,
@@ -164,7 +179,7 @@ func (consumer Consumer) startGoroutinesWithRetries(
 // binds the queue to the routing key(s), and starts the goroutines
 // that will consume from the queue
 func (consumer Consumer) startGoroutines(
-	handler func(d Delivery) bool,
+	handler Handler,
 	queue string,
 	routingKeys []string,
 	consumeOptions ConsumeOptions,
@@ -246,12 +261,18 @@ func (consumer Consumer) startGoroutines(
 					handler(Delivery{msg})
 					continue
 				}
-				if handler(Delivery{msg}) {
+				switch handler(Delivery{msg}) {
+				case Ack:
 					err := msg.Ack(false)
 					if err != nil {
 						consumer.logger.Printf("can't ack message: %v", err)
 					}
-				} else {
+				case NackDiscard:
+					err := msg.Nack(false, false)
+					if err != nil {
+						consumer.logger.Printf("can't nack message: %v", err)
+					}
+				case NackRequeue:
 					err := msg.Nack(false, true)
 					if err != nil {
 						consumer.logger.Printf("can't nack message: %v", err)

examples/consumer/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"log"
 
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 	rabbitmq "github.com/wagslane/go-rabbitmq"
 )
 
@@ -16,10 +16,9 @@ func main() {
 		log.Fatal(err)
 	}
 	err = consumer.StartConsuming(
-		func(d rabbitmq.Delivery) bool {
+		func(d rabbitmq.Delivery) (action rabbitmq.Action) {
 			log.Printf("consumed: %v", string(d.Body))
-			// true to ACK, false to NACK
-			return true
+			return
 		},
 		"my_queue",
 		[]string{"routing_key", "routing_key_2"},

examples/logger/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"log"
 
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 	rabbitmq "github.com/wagslane/go-rabbitmq"
 )
 

examples/publisher/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"log"
 
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 	rabbitmq "github.com/wagslane/go-rabbitmq"
 )
 

go.mod

@@ -2,4 +2,4 @@ module github.com/wagslane/go-rabbitmq
 
 go 1.16
 
-require github.com/streadway/amqp v1.0.0
+require github.com/rabbitmq/amqp091-go v0.0.0-20210823000215-c428a6150891

go.sum

@@ -1,2 +1,2 @@
-github.com/streadway/amqp v1.0.0 h1:kuuDrUJFZL1QYL9hUNuCxNObNzB0bV/ZG5jV3RWAQgo=
-github.com/streadway/amqp v1.0.0/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
+github.com/rabbitmq/amqp091-go v0.0.0-20210823000215-c428a6150891 h1:13nv5f/LNJxNpvpYm/u0NqrlFebon342f9Xu9GpklKc=
+github.com/rabbitmq/amqp091-go v0.0.0-20210823000215-c428a6150891/go.mod h1:ogQDLSOACsLPsIq0NpbtiifNZi2YOz0VTJ0kHRghqbM=

publish.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 // DeliveryMode. Transient means higher throughput but messages will not be

table.go

@@ -1,6 +1,6 @@
 package rabbitmq
 
-import "github.com/streadway/amqp"
+import amqp "github.com/rabbitmq/amqp091-go"
 
 // Table stores user supplied fields of the following types:
 //

vendor/github.com/rabbitmq/amqp091-go/CODE_OF_CONDUCT.md

@@ -0,0 +1,77 @@
+# Contributor Covenant Code of Conduct
+
+## Our Pledge
+
+In the interest of fostering an open and welcoming environment, we as
+contributors and maintainers pledge to making participation in RabbitMQ Operator project and
+our community a harassment-free experience for everyone, regardless of age, body
+size, disability, ethnicity, sex characteristics, gender identity and expression,
+level of experience, education, socio-economic status, nationality, personal
+appearance, race, religion, or sexual identity and orientation.
+
+## Our Standards
+
+Examples of behavior that contributes to creating a positive environment
+include:
+
+* Using welcoming and inclusive language
+* Being respectful of differing viewpoints and experiences
+* Gracefully accepting constructive criticism
+* Focusing on what is best for the community
+* Showing empathy towards other community members
+
+Examples of unacceptable behavior by participants include:
+
+* The use of sexualized language or imagery and unwelcome sexual attention or
+  advances
+* Trolling, insulting/derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or electronic
+  address, without explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+  professional setting
+
+## Our Responsibilities
+
+Project maintainers are responsible for clarifying the standards of acceptable
+behavior and are expected to take appropriate and fair corrective action in
+response to any instances of unacceptable behavior.
+
+Project maintainers have the right and responsibility to remove, edit, or
+reject comments, commits, code, wiki edits, issues, and other contributions
+that are not aligned to this Code of Conduct, or to ban temporarily or
+permanently any contributor for other behaviors that they deem inappropriate,
+threatening, offensive, or harmful.
+
+## Scope
+
+This Code of Conduct applies both within project spaces and in public spaces
+when an individual is representing the project or its community. Examples of
+representing a project or community include using an official project e-mail
+address, posting via an official social media account, or acting as an appointed
+representative at an online or offline event. Representation of a project may be
+further defined and clarified by project maintainers.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported by contacting the project team at oss-coc@vmware.com. All
+complaints will be reviewed and investigated and will result in a response that
+is deemed necessary and appropriate to the circumstances. The project team is
+obligated to maintain confidentiality with regard to the reporter of an incident.
+Further details of specific enforcement policies may be posted separately.
+
+Project maintainers who do not follow or enforce the Code of Conduct in good
+faith may face temporary or permanent repercussions as determined by other
+members of the project's leadership.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
+available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html
+
+[homepage]: https://www.contributor-covenant.org
+
+For answers to common questions about this code of conduct, see
+https://www.contributor-covenant.org/faq
+

vendor/github.com/rabbitmq/amqp091-go/Makefile

@@ -0,0 +1,17 @@
+.DEFAULT_GOAL := list
+
+# Insert a comment starting with '##' after a target, and it will be printed by 'make' and 'make list'
+list: ## list Makefile targets
+	@echo "The most used targets: \n"
+	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}'
+
+
+fmt: ## Run go fmt against code
+	go fmt ./...
+
+
+vet: ## Run go vet against code
+	go vet ./...
+
+tests: ## Run all tests and requires a running rabbitmq-server
+	env AMQP_URL=amqp://guest:guest@127.0.0.1:5672/ go test -v -tags integration

vendor/github.com/rabbitmq/amqp091-go/go.mod

@@ -0,0 +1,3 @@
+module github.com/rabbitmq/amqp091-go
+
+go 1.15

vendor/github.com/streadway/amqp/go.mod

@@ -1,3 +0,0 @@
-module github.com/streadway/amqp
-
-go 1.10

vendor/modules.txt

@@ -1,3 +1,3 @@
-# github.com/streadway/amqp v1.0.0
+# github.com/rabbitmq/amqp091-go v0.0.0-20210823000215-c428a6150891
 ## explicit
-github.com/streadway/amqp
+github.com/rabbitmq/amqp091-go