Clarification on how direct exchange work

[kotyara85]

Found a somehow weird topics behavior. Go examples with empty queue name work just fine when it just creates a new tmp queue but here's my scenario
I do have one queue, let's call it messaging
That queue has multiple bindings:
bind.super
bind.super1
bind.super2
wrong.ok

I bind and consume this way:

	failOnError(ch.QueueBind(q.Name, "bind.super1", "exchagnge", false, nil), "Failed to bind")

	msgs, err := ch.Consume(
		"messaging",                               // queue
		"",                                   // consumer
		true,                                 // auto-ack
		false,                                // exclusive
		false,                                // no-local
		false,                                // no-wait
		nil, // args
	)

What is weird, rounting works just fine with bind.*, but if wrong.ok is being consumed by everyone ignoring bind.xyz. Is it expected behavior and is there a way to filter those out besides some extra go code?
Thanks!

[lukebakken]

I'm having a hard time following your description. Please provide a script I can run that demonstrates the behavior you describe. This script should do the following:

• Declare exchanges and queues
• Start a consumer
• Publish a message demonstrating the issue

[kotyara85]

I don't have any scripts, everything create in the ui.
To clarify.
I have one queue
I have one direct exchange with two queue bindings:
bind.super1 -> queue
bind.super2 -> queue

Consumers which are listening on queue with bind.super1 are able to see bind.super2 messages

Consumer script:

conn, err := amqp.Dial("amqps://admin:password@rmq:443")
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()

ch, err := conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()

q, err := ch.QueueDeclare(
	"queue", // name
	true,  // durable
	false, // delete when unused
	false, // exclusive
	false, // no-wait
	nil,   // arguments
)
failOnError(err, "Failed to declare a queue")

failOnError(ch.QueueBind(q.Name, "bind.super2", "direct", false, nil), "Failed to bind")

msgs, err := ch.Consume(
	q.Name, // queue
	"",     // consumer
	true,   // auto-ack
	false,  // exclusive
	false,  // no-local
	false,  // no-wait
	nil,    // args
)
failOnError(err, "Failed to register a consumer")

forever := make(chan bool)

go func() {
	for d := range msgs {

		log.Printf("Received a message: %s, %s", d.Body, d.RoutingKey)
	}
}()

log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever

Output:

2021/11/12 09:40:45  [*] Waiting for messages. To exit press CTRL+C
2021/11/12 09:40:52 Received a message: Hello World!, bind.super1

[lukebakken]

I don't see anything out of the ordinary which is why I need an automated way to reproduce your environment. We don't have time to guess how to reproduce something like this.

You can use the rabbitmqadmin and other tools to set up the exact same environment. Provide the Go code for your producer and consumer. If I can reproduce this I can tell you what is incorrect. More than likely it's a small misconfiguration.

[kotyara85]

Here it's

rabbitmqadmin declare exchange --vhost=/ name=direct type=direct
rabbitmqadmin declare queue --vhost=/ name=queue
rabbitmqadmin declare binding --vhost=/ source="direct" destination_type="queue" destination="queue" routing_key="bind.super1"
rabbitmqadmin declare binding --vhost=/ source="direct" destination_type="queue" destination="queue" routing_key="bind.super2"

consumer1

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"queue", // name
		true,    // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a queue")

	failOnError(ch.QueueBind(q.Name, "bind.super1", "direct", false, nil), "Failed to bind")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s, %s", d.Body, d.RoutingKey)
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

consumer2

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"queue", // name
		true,    // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a queue")

	failOnError(ch.QueueBind(q.Name, "bind.super2", "direct", false, nil), "Failed to bind")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s, %s", d.Body, d.RoutingKey)
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

producer

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	body := "Hello World!"
	err = ch.Publish(
		"direct",      // exchange
		"bind.super2", // routing key
		false,         // mandatory
		false,         // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
			Headers:     nil,
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)
}

So with this config bind.super2 delivered to both consumers

[mkuratczyk]

You bind the same queue "queue" to the direct exchange "direct", using two routing keys. That means that messages with the key "bind.super1" and "bind.super2" will be delivered to the queue "queue" and messages with any other keys will be dropped/rejected (since they won't match any bindings). If you want different clients to receive only messages with one routing key, you need two queues, each bound with a different key. Then each consumer can consume from one of those queues.

[lukebakken]

As @mkuratczyk said, bindings are per-queue, not per-consumer.