Extension

The x-ruststream-amqp extension lists each binding of a subscription: exchange, type, routing key and arguments, in declaration order.

The specification's channel binding is a queue or a routing key, and it has no field for what joins the two. So a subscription reports its bindings in this crate's extension, beside the amqp binding of its channel. The extension holds a bindings list with one entry per bind and bind_with, in the order the descriptor declares them.

An entry carries the exchange name and its type: direct, fanout, headers, topic, or the type a plugin exchange declares. The routingKey is written even when it is empty, because an empty key is the binding key of a fanout or a headers exchange. The arguments appear when the binding has any, each value in the JSON shape of its AMQP type. A subscription with no binding has no extension. An extension is not a binding, so it carries no bindingVersion.

Example

The queue inventory.events is bound to a topic exchange by its key and to a headers exchange by its arguments.

Macros

examples/lapin_asyncapi.rs

/// What a headers exchange matches a binding on: every entry, or any one of them.
fn gold_tier() -> FieldTable {
    let mut arguments = FieldTable::default();
    arguments.insert("x-match".into(), AMQPValue::LongString("all".into()));
    arguments.insert("tier".into(), AMQPValue::LongString("gold".into()));
    arguments
}

#[subscriber(
    RabbitQueue::new("inventory.events")
        .bind(RabbitExchange::topic("events"), "inventory.*")
        .bind_with(RabbitExchange::headers("attributes"), "", gold_tier())
)]
async fn on_event(event: &Event, Out(restocks): Out<impl Publisher, Restocks>) -> HandlerOutcome {
    if event.in_stock > 0 {
        return HandlerOutcome::ack();
    }
    let restock = Restock {
        sku: event.sku.clone(),
    };
    if restocks.message(&restock).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/lapin_asyncapi.rs

/// What a headers exchange matches a binding on: every entry, or any one of them.
fn gold_tier() -> FieldTable {
    let mut arguments = FieldTable::default();
    arguments.insert("x-match".into(), AMQPValue::LongString("all".into()));
    arguments.insert("tier".into(), AMQPValue::LongString("gold".into()));
    arguments
}

/// The events queue, fed by a topic exchange and by a headers exchange.
fn events() -> RabbitQueue {
    RabbitQueue::new("inventory.events")
        .bind(RabbitExchange::topic("events"), "inventory.*")
        .bind_with(RabbitExchange::headers("attributes"), "", gold_tier())
}

The service's document, in excerpt: the extension sits under the channel inventory.events. A test holds the service to every value of the excerpt.

tests/snapshots/asyncapi-amqp.json

{
  "servers": {
    "rabbit": {
      "host": "rabbit:5672",
      "protocol": "amqp",
      "protocolVersion": "0.9.1"
    }
  },
  "channels": {
    "inventory.check": {
      "address": "inventory.check",
      "bindings": {
        "amqp": {
          "is": "queue",
          "queue": {
            "name": "inventory.check",
            "durable": false,
            "exclusive": true,
            "autoDelete": true
          },
          "bindingVersion": "0.3.0"
        }
      }
    },
    "inventory.events": {
      "address": "inventory.events",
      "bindings": {
        "amqp": {
          "is": "queue",
          "queue": {
            "name": "inventory.events"
          },
          "bindingVersion": "0.3.0"
        },
        "x-ruststream-amqp": {
          "bindings": [
            {
              "exchange": "events",
              "type": "topic",
              "routingKey": "inventory.*"
            },
            {
              "exchange": "attributes",
              "type": "headers",
              "routingKey": "",
              "arguments": {
                "x-match": "all",
                "tier": "gold"
              }
            }
          ]
        }
      }
    },
    "inventory.unrouted": {
      "address": null,
      "bindings": {
        "amqp": {
          "is": "routingKey",
          "exchange": {
            "name": "answers"
          },
          "bindingVersion": "0.3.0"
        }
      }
    },
    "inventory.dead": {
      "address": "inventory.dead",
      "bindings": {
        "amqp": {
          "is": "queue",
          "queue": {
            "name": "inventory.dead"
          },
          "bindingVersion": "0.3.0"
        }
      }
    }
  },
  "operations": {
    "receive_inventory_check": {
      "action": "receive",
      "bindings": {
        "amqp": {
          "ack": true,
          "bindingVersion": "0.3.0"
        }
      },
      "reply": {
        "address": {
          "location": "$message.header#/reply-to"
        }
      }
    },
    "send_inventory_check_inventory_dead": {
      "action": "send",
      "bindings": {
        "amqp": {
          "deliveryMode": 2,
          "bindingVersion": "0.3.0"
        }
      }
    }
  }
}