# Routing (Direct) (/connectors/rabbitmq/how-to/routing)



**Routing** delivers a message **selectively** — only to queues whose binding key **exactly** matches the message's routing key. In AMQP this is a **direct** exchange. It sits between fanout (everyone) and topics (pattern matching). The direct exchange is **virtual connector-side routing**: at publish time the connector matches the routing key against the bindings and writes a copy to each matched queue's KubeMQ channel.

## Overview [#overview]

Declare a **direct** exchange and bind queues with specific keys — for example `info` → an info queue, `error` → an error queue. Publishing with a routing key delivers the message **only** to queues bound on that exact key. Multiple bindings on the **same** key all match (a key can fan out to several queues). A routing key with **no** matching binding is **silently dropped** unless you set `mandatory=true`, which returns a `312 NO_ROUTE`.

| Operation        | AMQP action                                                  | KubeMQ mapping                                  |
| ---------------- | ------------------------------------------------------------ | ----------------------------------------------- |
| Declare exchange | `exchange.declare("direct_logs", "direct")`                  | Virtual direct routing entry (no storage)       |
| Bind             | `queue.bind(q, "direct_logs", "error")`                      | Maps key `error` → queue channel                |
| Publish          | `basic.publish(exchange="direct_logs", routing-key="error")` | Copy written to each queue bound on `error`     |
| Unmatched key    | no binding on the key                                        | Silently dropped (or `312` if `mandatory=true`) |

## How it works [#how-it-works]

A publish resolves to exactly the queues bound on the message's routing key. A key bound to nobody routes to nobody and is dropped without error.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher&#x22;]
CONN[&#x22;RabbitMQ connector<br/>:5672&#x22;]
X([&#x22;direct exchange<br/>direct_logs (virtual)&#x22;])
BROKER[&#x22;Message Broker&#x22;]
QE{{&#x22;error queue&#x22;}}
QI{{&#x22;info queue&#x22;}}
CE[&#x22;error consumer&#x22;]
CI[&#x22;info consumer&#x22;]

PUB -- &#x22;key=error&#x22; --> CONN
PUB -- &#x22;key=info&#x22; --> CONN
PUB -. &#x22;key=debug (unbound)&#x22; .-> CONN
CONN --> X
X -- &#x22;match error&#x22; --> QE
X -- &#x22;match info&#x22; --> QI
X -. &#x22;debug → no binding → dropped&#x22; .-> BROKER
QE --> BROKER
QI --> BROKER
BROKER --> CE
BROKER --> CI

class PUB,CE,CI client
class CONN connector
class X queue
class BROKER broker
class QE,QI queue
`"
/>

*The virtual direct exchange routes each publish only to queues bound on the exact routing key; `error` reaches the error queue, `info` the info queue, and an unbound `debug` key is silently dropped.*

## Publish and route [#publish-and-route]

Each example declares a direct exchange `direct_logs`, binds one consumer on `error` and another on `info` (each via its own server-named exclusive queue), then publishes three keys: `error`, `info`, and `debug`. The `error` and `info` messages reach their respective consumers; `debug` is bound to nobody and is silently dropped. Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://guest:guest@localhost:5672/`).

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"log"
    	"os"
    	"time"

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

    const exchange = "direct_logs"

    func amqpURL() string {
    	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
    		return v
    	}
    	return "amqp://guest:guest@localhost:5672/"
    }

    func bindConsumer(conn *amqp.Connection, key string) <-chan amqp.Delivery {
    	ch, err := conn.Channel()
    	if err != nil {
    		log.Fatalf("channel for %s: %v", key, err)
    	}
    	q, err := ch.QueueDeclare("", false, false, true, false, nil) // server-named, exclusive
    	if err != nil {
    		log.Fatalf("declare %s: %v", key, err)
    	}
    	if err := ch.QueueBind(q.Name, key, exchange, false, nil); err != nil { // bind on the exact key
    		log.Fatalf("bind %s: %v", key, err)
    	}
    	msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
    	if err != nil {
    		log.Fatalf("consume %s: %v", key, err)
    	}
    	return msgs
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
    	defer cancel()

    	conn, err := amqp.Dial(amqpURL())
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	defer func() { _ = conn.Close() }()

    	ch, _ := conn.Channel()
    	if err := ch.ExchangeDeclare(exchange, "direct", false, false, false, false, nil); err != nil {
    		log.Fatalf("declare exchange: %v", err)
    	}

    	errorMsgs := bindConsumer(conn, "error")
    	infoMsgs := bindConsumer(conn, "info")

    	for key, body := range map[string]string{
    		"error": "an error happened",
    		"info":  "all is well",
    		"debug": "nobody is bound to debug", // silently dropped (no binding)
    	} {
    		if err := ch.PublishWithContext(ctx, exchange, key, false, false, amqp.Publishing{
    			ContentType: "text/plain",
    			Body:        []byte(body),
    		}); err != nil {
    			log.Fatalf("publish key=%s: %v", key, err)
    		}
    	}
    	log.Printf(" [x] Published keys: error, info, debug")

    	select {
    	case d := <-errorMsgs:
    		log.Printf(" [error] received %q (key=%q)", d.Body, d.RoutingKey)
    	case <-ctx.Done():
    		log.Fatal("error consumer timed out")
    	}
    	select {
    	case d := <-infoMsgs:
    		log.Printf(" [info]  received %q (key=%q)", d.Body, d.RoutingKey)
    	case <-ctx.Done():
    		log.Fatal("info consumer timed out")
    	}
    	log.Printf(" [✓] debug (unbound) → silently dropped (no consumer)")
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    import pika

    URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
    EXCHANGE = "direct_logs"


    def main() -> None:
        conn = pika.BlockingConnection(pika.URLParameters(URL))
        ch = conn.channel()
        ch.exchange_declare(exchange=EXCHANGE, exchange_type="direct", durable=False)

        consumers = {}
        for key in ("error", "info"):
            queue = ch.queue_declare(queue="", exclusive=True).method.queue
            ch.queue_bind(exchange=EXCHANGE, queue=queue, routing_key=key)  # bind on the exact key
            consumers[key] = queue

        for key, body in {
            "error": "an error happened",
            "info": "all is well",
            "debug": "nobody is bound to debug",  # silently dropped (no binding)
        }.items():
            ch.basic_publish(exchange=EXCHANGE, routing_key=key, body=body.encode())
        print(" [x] Published keys: error, info, debug")

        for key, queue in consumers.items():
            for method, _props, body in ch.consume(queue, inactivity_timeout=10, auto_ack=True):
                if method is None:
                    raise SystemExit(f"{key} consumer timed out")
                print(f" [{key}] received {body.decode()!r} (key={method.routing_key})")
                break
            ch.cancel()

        print(" [✓] debug (unbound) → silently dropped (no consumer)")
        ch.close()
        conn.close()


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import java.nio.charset.StandardCharsets;
    import java.util.LinkedHashMap;
    import java.util.Map;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;

    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;

    public final class Main {
        private static final String EXCHANGE = "direct_logs";

        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"));
            if (factory.getVirtualHost() == null || factory.getVirtualHost().isEmpty()) {
                factory.setVirtualHost("/");
            }

            try (Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel()) {
                channel.exchangeDeclare(EXCHANGE, "direct", false);

                Map<String, BlockingQueue<String>> received = new LinkedHashMap<>();
                for (String key : new String[] {"error", "info"}) {
                    Channel ch = connection.createChannel();
                    String queue = ch.queueDeclare("", false, true, true, null).getQueue();
                    ch.queueBind(queue, EXCHANGE, key); // bind on the exact key
                    BlockingQueue<String> sink = new ArrayBlockingQueue<>(4);
                    ch.basicConsume(queue, true,
                            (tag, d) -> sink.offer(new String(d.getBody(), StandardCharsets.UTF_8)),
                            tag -> { });
                    received.put(key, sink);
                }

                Map<String, String> publishes = new LinkedHashMap<>();
                publishes.put("error", "an error happened");
                publishes.put("info", "all is well");
                publishes.put("debug", "nobody is bound to debug"); // silently dropped
                for (Map.Entry<String, String> e : publishes.entrySet()) {
                    channel.basicPublish(EXCHANGE, e.getKey(), null,
                            e.getValue().getBytes(StandardCharsets.UTF_8));
                }
                System.out.println("[x] Published keys: error, info, debug");

                for (Map.Entry<String, BlockingQueue<String>> e : received.entrySet()) {
                    String body = e.getValue().poll(10, TimeUnit.SECONDS);
                    if (body == null) throw new IllegalStateException(e.getKey() + " consumer timed out");
                    System.out.println("[" + e.getKey() + "] received " + body);
                }
                System.out.println("[v] debug (unbound) → silently dropped (no consumer)");
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import amqp, { type Channel } from "amqplib";

    const EXCHANGE = "direct_logs";

    function url(): string {
      return process.env["KUBEMQ_AMQP_URL"] ?? "amqp://guest:guest@localhost:5672/";
    }

    function once(ch: Channel, queue: string): Promise<string> {
      return new Promise((resolve) => {
        ch.consume(queue, (msg) => {
          if (msg) resolve(msg.content.toString());
        }, { noAck: true });
      });
    }

    async function main(): Promise<void> {
      const connection = await amqp.connect(url());
      const channel = await connection.createChannel();
      await channel.assertExchange(EXCHANGE, "direct", { durable: false });

      const consumers: Record<string, Promise<string>> = {};
      for (const key of ["error", "info"]) {
        const ch = await connection.createChannel();
        const q = await ch.assertQueue("", { exclusive: true });
        await ch.bindQueue(q.queue, EXCHANGE, key); // bind on the exact key
        consumers[key] = once(ch, q.queue);
      }

      const publishes: Record<string, string> = {
        error: "an error happened",
        info: "all is well",
        debug: "nobody is bound to debug", // silently dropped (no binding)
      };
      for (const [key, body] of Object.entries(publishes)) {
        channel.publish(EXCHANGE, key, Buffer.from(body));
      }
      console.log("[x] Published keys: error, info, debug");

      console.log(`[error] received ${await consumers["error"]}`);
      console.log(`[info]  received ${await consumers["info"]}`);
      console.log("[v] debug (unbound) → silently dropped (no consumer)");
      await connection.close();
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Collections.Concurrent;
    using System.Text;
    using RabbitMQ.Client;
    using RabbitMQ.Client.Events;

    const string exchange = "direct_logs";

    static string Url() =>
        Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
            ? v
            : "amqp://guest:guest@localhost:5672/";

    var factory = new ConnectionFactory { Uri = new Uri(Url()) };
    await using var connection = await factory.CreateConnectionAsync("routing-direct");
    await using var channel = await connection.CreateChannelAsync();
    await channel.ExchangeDeclareAsync(exchange, ExchangeType.Direct, durable: false);

    var received = new Dictionary<string, BlockingCollection<string>>();
    foreach (var key in new[] { "error", "info" })
    {
        var ch = await connection.CreateChannelAsync();
        var queue = (await ch.QueueDeclareAsync("", durable: false, exclusive: true, autoDelete: true)).QueueName;
        await ch.QueueBindAsync(queue, exchange, key); // bind on the exact key
        var sink = new BlockingCollection<string>();
        var consumer = new AsyncEventingBasicConsumer(ch);
        consumer.ReceivedAsync += (_, ea) =>
        {
            sink.Add(Encoding.UTF8.GetString(ea.Body.Span));
            return Task.CompletedTask;
        };
        await ch.BasicConsumeAsync(queue, autoAck: true, consumer: consumer);
        received[key] = sink;
    }

    var publishes = new Dictionary<string, string>
    {
        ["error"] = "an error happened",
        ["info"] = "all is well",
        ["debug"] = "nobody is bound to debug", // silently dropped (no binding)
    };
    foreach (var (key, body) in publishes)
        await channel.BasicPublishAsync(exchange, key, body: Encoding.UTF8.GetBytes(body));
    Console.WriteLine("[x] Published keys: error, info, debug");

    foreach (var (key, sink) in received)
        Console.WriteLine($"[{key}] received {sink.Take()}");
    Console.WriteLine("[v] debug (unbound) → silently dropped (no consumer)");
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # frozen_string_literal: true
    require "bunny"
    require "amq/uri"

    EXCHANGE = "direct_logs"

    opts = AMQ::URI.parse(ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"))
    opts[:vhost] = "/" if opts[:vhost].nil? || opts[:vhost].to_s.empty?
    conn = Bunny.new(opts)
    conn.start

    ch = conn.create_channel
    exchange = ch.direct(EXCHANGE, durable: false)

    consumers = {}
    %w[error info].each do |key|
      sub_ch = conn.create_channel
      queue = sub_ch.queue("", exclusive: true)
      queue.bind(exchange, routing_key: key) # bind on the exact key
      sink = Queue.new
      queue.subscribe(manual_ack: false, block: false) { |_di, _props, body| sink.push(body) }
      consumers[key] = sink
    end

    {
      "error" => "an error happened",
      "info"  => "all is well",
      "debug" => "nobody is bound to debug" # silently dropped (no binding)
    }.each { |key, body| exchange.publish(body, routing_key: key) }
    puts " [x] Published keys: error, info, debug"

    consumers.each { |key, sink| puts " [#{key}] received #{sink.pop.inspect}" }
    puts " [x] debug (unbound) → silently dropped (no consumer)"

    conn.close
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use futures_lite::StreamExt;
    use lapin::{
        options::{
            BasicConsumeOptions, BasicPublishOptions, ExchangeDeclareOptions, QueueBindOptions,
            QueueDeclareOptions,
        },
        types::FieldTable,
        BasicProperties, Connection, ConnectionProperties, ExchangeKind,
    };

    const EXCHANGE: &str = "direct_logs";

    fn amqp_url() -> String {
        let url = std::env::var("KUBEMQ_AMQP_URL")
            .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
        match url.rsplit_once('@').map_or(url.as_str(), |(_, host)| host) {
            host if host.ends_with('/') && !host.to_lowercase().ends_with("/%2f") => format!("{url}%2f"),
            _ => url,
        }
    }

    async fn bind_consumer(conn: &Connection, key: &str) -> Result<lapin::Consumer, Box<dyn std::error::Error>> {
        let ch = conn.create_channel().await?;
        let queue = ch
            .queue_declare("", QueueDeclareOptions { exclusive: true, ..Default::default() }, FieldTable::default())
            .await?;
        ch.queue_bind(queue.name().as_str(), EXCHANGE, key, QueueBindOptions::default(), FieldTable::default())
            .await?; // bind on the exact key
        Ok(ch
            .basic_consume(queue.name().as_str(), "", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
            .await?)
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let conn = Connection::connect(&amqp_url(), ConnectionProperties::default()).await?;
        let ch = conn.create_channel().await?;
        ch.exchange_declare(EXCHANGE, ExchangeKind::Direct, ExchangeDeclareOptions::default(), FieldTable::default())
            .await?;

        let mut error_c = bind_consumer(&conn, "error").await?;
        let mut info_c = bind_consumer(&conn, "info").await?;

        for (key, body) in [
            ("error", "an error happened"),
            ("info", "all is well"),
            ("debug", "nobody is bound to debug"), // silently dropped (no binding)
        ] {
            ch.basic_publish(EXCHANGE, key, BasicPublishOptions::default(), body.as_bytes(), BasicProperties::default())
                .await?;
        }
        println!("[x] Published keys: error, info, debug");

        let err = error_c.next().await.ok_or("error consumer closed")??;
        println!("[error] received {}", String::from_utf8_lossy(&err.data));
        let info = info_c.next().await.ok_or("info consumer closed")??;
        println!("[info]  received {}", String::from_utf8_lossy(&info.data));
        println!("[x] debug (unbound) → silently dropped (no consumer)");

        conn.close(0, "done").await?;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Unmatched key — silent drop [#unmatched-key--silent-drop]

This is the behavior to internalize: an unroutable publish **without** `mandatory` produces **no error and no message**. The publish succeeds at the protocol level, but the connector resolves it to an empty set of queues and writes nothing. If you need to detect unroutable publishes, set `mandatory=true` and handle the returned `basic.return(312 NO_ROUTE)`.

<Callout type="info">
  **A key can fan out to several queues.** Multiple queues bound on the **same** key all receive a copy — direct routing is not limited to one queue per key. If you need wildcard or hierarchical keys, use a [topic exchange](/connectors/rabbitmq/how-to/topics) instead of binding a long list of exact keys.
</Callout>

## Related [#related]

<Cards>
  <Card title="Topics" href="/connectors/rabbitmq/how-to/topics" description="Pattern routing with * (one word) and # (zero or more words) over dot-separated keys." />

  <Card title="Pub/Sub (fanout)" href="/connectors/rabbitmq/how-to/pub-sub" description="Broadcast to all subscribers — the routing key is ignored." />

  <Card title="Channel mapping" href="/connectors/rabbitmq/reference/channel-mapping" description="How exchanges, bindings, and routing keys resolve to KubeMQ Queue channels." />
</Cards>