STOMP: handle declarations differently

mkuratczyk · mkuratczyk · commit 3f19cc13a044 · 2026-03-13T13:55:50.000+01:00
RMQ 4.3+ doesn't allow transient non-exclusive
queues by default so we had tests failing.

I've settled on this:
--queues arguments is honours, but declarations
happen through STOMP headers, rather than AMQP 1.0 management

if --queues=predeclared, --queue-durability=none and
consumption from /topic - declare exclusive queues.

For other /amq/queue/ destinations (which expect pre-declared
queues), declare them over AMQP 1.0 management.
diff --git a/pkg/mgmt/mgmt.go b/pkg/mgmt/mgmt.go
@@ -103,15 +103,15 @@ func (m *Mgmt) DeclareQueues(cfg config.Config) {
 			log.Info("Not declaring queues for AMQP consumers since the address doesn't start with /queues/")
 		}
 	}
-	// declare queues for STOMP publishers
+	// declare queues for STOMP publishers using /amq/queue/ destination
 	if cfg.PublisherProto == config.STOMP && strings.HasPrefix(cfg.PublishTo, "/amq/queue/") {
 		for i := 1; i <= cfg.Publishers; i++ {
 			queueName := utils.ResolveTerminus(cfg.PublishToTemplate, i)
 			queueName = strings.TrimPrefix(queueName, "/amq/queue/")
 			m.DeclareAndBind(cfg, queueName, i)
 		}
 	}
-	// declare queues for STOMP consumers
+	// declare queues for STOMP consumers using /amq/queue/ destination
 	if cfg.ConsumerProto == config.STOMP && strings.HasPrefix(cfg.ConsumeFrom, "/amq/queue/") {
 		for i := 1; i <= cfg.Consumers; i++ {
 			q := utils.ResolveTerminus(cfg.ConsumeFromTemplate, i)
diff --git a/pkg/stomp/consumer.go b/pkg/stomp/consumer.go
@@ -115,7 +115,7 @@ func (c *StompConsumer) Subscribe() {
 	var err error
 
 	if c.Connection != nil {
-		sub, err = c.Connection.Subscribe(c.Topic, stomp.AckClient, buildSubscribeOpts(c.Config)...)
+		sub, err = c.Connection.Subscribe(c.Topic, stomp.AckClient, buildSubscribeOpts(c.Config, c.Topic)...)
 		if err != nil {
 			log.Error("subscription failed", "id", c.Id, "queue", c.Topic, "error", err.Error())
 			return
@@ -215,7 +215,7 @@ func (c *StompConsumer) Stop(reason string) {
 	log.Debug("consumer stopped", "id", c.Id, "reason", reason)
 }
 
-func buildSubscribeOpts(cfg config.Config) []func(*frame.Frame) error {
+func buildSubscribeOpts(cfg config.Config, destination string) []func(*frame.Frame) error {
 	var subscribeOpts []func(*frame.Frame) error
 
 	subscribeOpts = append(subscribeOpts,
@@ -238,6 +238,29 @@ func buildSubscribeOpts(cfg config.Config) []func(*frame.Frame) error {
 		subscribeOpts = append(subscribeOpts,
 			stomp.SubscribeOpt.Header("x-stream-filter", cfg.StreamFilterValues))
 	}
+
+	switch cfg.Queues {
+	case config.Classic:
+		subscribeOpts = append(subscribeOpts,
+			stomp.SubscribeOpt.Header("x-queue-type", "classic"))
+	case config.Quorum:
+		subscribeOpts = append(subscribeOpts,
+			stomp.SubscribeOpt.Header("x-queue-type", "quorum"))
+	case config.Stream:
+		subscribeOpts = append(subscribeOpts,
+			stomp.SubscribeOpt.Header("x-queue-type", "stream"))
+	case config.Exclusive:
+		subscribeOpts = append(subscribeOpts,
+			stomp.SubscribeOpt.Header("exclusive", "true"))
+	case config.Predeclared:
+		// For /topic/ destinations with transient queues (QueueDurability == None),
+		// use exclusive to avoid deprecated transient non-exclusive queues
+		if strings.HasPrefix(destination, "/topic/") && cfg.QueueDurability == config.None {
+			subscribeOpts = append(subscribeOpts,
+				stomp.SubscribeOpt.Header("exclusive", "true"))
+		}
+	}
+
 	return subscribeOpts
 }
 

Original file line number	Diff line number	Diff line change
`@@ -103,15 +103,15 @@ func (m *Mgmt) DeclareQueues(cfg config.Config) {`
`103`	`103`	`log.Info("Not declaring queues for AMQP consumers since the address doesn't start with /queues/")`
`104`	`104`	`}`
`105`	`105`	`}`
`106`		`- // declare queues for STOMP publishers`
	`106`	`+ // declare queues for STOMP publishers using /amq/queue/ destination`
`107`	`107`	`if cfg.PublisherProto == config.STOMP && strings.HasPrefix(cfg.PublishTo, "/amq/queue/") {`
`108`	`108`	`for i := 1; i <= cfg.Publishers; i++ {`
`109`	`109`	`queueName := utils.ResolveTerminus(cfg.PublishToTemplate, i)`
`110`	`110`	`queueName = strings.TrimPrefix(queueName, "/amq/queue/")`
`111`	`111`	`m.DeclareAndBind(cfg, queueName, i)`
`112`	`112`	`}`
`113`	`113`	`}`
`114`		`- // declare queues for STOMP consumers`
	`114`	`+ // declare queues for STOMP consumers using /amq/queue/ destination`
`115`	`115`	`if cfg.ConsumerProto == config.STOMP && strings.HasPrefix(cfg.ConsumeFrom, "/amq/queue/") {`
`116`	`116`	`for i := 1; i <= cfg.Consumers; i++ {`
`117`	`117`	`q := utils.ResolveTerminus(cfg.ConsumeFromTemplate, i)`