Add support for exclusive queues (amqp091 only for now)

Author: mkuratczyk

cmd/root.go

@@ -493,6 +493,11 @@ func start(cfg config.Config) {
 		log.Debug("will stop all consumers and publishers at " + time.Now().Add(cfg.Duration).String())
 	}
 
+	if cfg.Queues == config.Exclusive && cfg.ConsumerProto != config.AMQP091 {
+		log.Error("exclusive queues are only supported for AMQP 0.9.1 consumers")
+		os.Exit(1)
+	}
+
 	// every  second, print the current values of the metrics
 	wg.Wait()
 }

main_test.go

@@ -649,6 +649,28 @@ var _ = Describe("OMQ CLI", func() {
 			Expect(outputStr).To(ContainSubstring("enabled:true"))
 		})
 	})
+
+	Describe("supports exclusive queues", func() {
+		It("declares exclusive queues for AMQP 0.9.1 consumers", func() {
+			args := []string{
+				"amqp091",
+				"-y", "1",
+				"-C", "0", // run indefinitely
+				"-T", "/queues/exclusive-amqp091",
+				"--queues", "exclusive",
+			}
+
+			rmqc, err := rabbithole.NewClient("http://127.0.0.1:15672", "guest", "guest")
+			Expect(err).ShouldNot(HaveOccurred())
+			session := omq(args)
+			defer session.Kill()
+
+			Eventually(func() bool {
+				q, err := rmqc.GetQueue("/", "exclusive-amqp091")
+				return err == nil && q.Name == "exclusive-amqp091" && q.OwnerPidDetails.Name != ""
+			}).WithTimeout(5 * time.Second).Should(BeTrue())
+		})
+	})
 })
 
 func omq(args []string) *gexec.Session {

pkg/amqp091/consumer.go

@@ -101,6 +101,23 @@ func (c *Amqp091Consumer) Connect() {
 func (c *Amqp091Consumer) Subscribe() {
 	if c.Connection != nil {
 		_ = c.Channel.Qos(c.Config.ConsumerCredits, 0, false)
+
+		if c.Config.Queues == config.Exclusive {
+			queueName := strings.TrimPrefix(c.Terminus, "/queues/")
+			_, err := c.Channel.QueueDeclare(
+				queueName, // name
+				false,     // durable
+				false,     // autoDelete
+				true,      // exclusive
+				false,     // noWait
+				nil,       // args
+			)
+			if err != nil {
+				log.Error("failed to declare exclusive queue", "id", c.Id, "queue", queueName, "error", err.Error())
+				return
+			}
+		}
+
 		// TODO add auto-ack and exclusive options
 		consumeArgs := amqp091.Table{}
 		if c.Config.StreamOffset != "" {

pkg/config/config.go

@@ -33,6 +33,7 @@ const (
 	Classic
 	Quorum
 	Stream
+	Exclusive
 )
 
 var AmqpDurabilityModes = map[AmqpDurabilityMode][]string{
@@ -46,6 +47,7 @@ var QueueTypes = map[QueueType][]string{
 	Classic:     {"classic"},
 	Quorum:      {"quorum"},
 	Stream:      {"stream"},
+	Exclusive:   {"exclusive"},
 }
 
 type AmqpOptions struct {

pkg/mgmt/mgmt.go

@@ -113,7 +113,7 @@ func (m *Mgmt) DeclareQueues(cfg config.Config) {
 }
 
 func (m *Mgmt) DeclareAndBind(cfg config.Config, queueName string, id int) *rmq.AmqpQueueInfo {
-	if cfg.Queues == config.Predeclared || m.declaredQueues[queueName] {
+	if cfg.Queues == config.Predeclared || cfg.Queues == config.Exclusive || m.declaredQueues[queueName] {
 		return nil
 	}