AMQP bindings

With the asyncapi feature the generated document says AMQP 1.0: the protocol key amqp1, and the crate's facts under the x-ruststream-amqp1 extension.

The core builds the AsyncAPI document from the app, before anything connects. The specification has an amqp1 binding, and every one of its objects must stay empty. So this crate writes what it knows into one extension object, x-ruststream-amqp1, at each level where the binding would sit. A server names the protocol and the container the service presents. A channel names its node address and, for a subscription, the link it reads over. A send operation says how it posts. A reply routed per request says where its address is read.

Everything comes from the service's own declarations: the broker, the descriptors and the publish policies. The document is published and shared, so no binding carries a credential. The binary #[ruststream::app] writes prints the document with asyncapi gen.

Example

The dependency line:

ruststream-amqp = { version = "0.7", features = ["asyncapi"] }

The service registers its broker under the label amqp, the key of its server in the document. cargo run --example amqp_asyncapi --features asyncapi,transaction -- asyncapi gen prints the document.

Macros
Manual

examples/amqp_asyncapi.rs
fn service(broker: AmqpBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("billing", "1.0.0")).with_broker_labeled("amqp", broker, |b| {
        b.include(confirm).out_reply(Publish).transform(ToReplyTo);
        b.include(watch)
            .max_attempts(nonzero!(3u32))
            .dead_letter("orders.dead");
        b.include(audit);
        b.include(book).out(Ledger, TransactionalPublish).build();
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    let password = env::var("AMQP_PASSWORD").unwrap_or_default();
    service(broker(&password))
}

examples/manual/amqp_asyncapi.rs
fn service(broker: AmqpBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("billing", "1.0.0")).with_broker_labeled("amqp", broker, |b| {
        let confirm = subscriber(orders(), Confirm);
        b.include(confirm.reply().to("receipts").build())
            .out_reply(Publish)
            .transform(ToReplyTo);
        b.include(subscriber(events(), Watch).build())
            .max_attempts(nonzero!(3u32))
            .dead_letter("orders.dead");
        b.include(subscriber(audit_queue(), Audit).build());
        let book = subscriber(AmqpAddress::queue("payments"), Book);
        b.include(book.build())
            .out(Ledger, TransactionalPublish)
            .build();
    })
}

fn app() -> impl App<State = ()> {
    let password = env::var("AMQP_PASSWORD").unwrap_or_default();
    service(broker(&password))
}

Its server and the channel of orders in the document:

tests/asyncapi_excerpt.json
{
  "servers": {
    "amqp": {
      "host": "broker.example.com:5672",
      "protocol": "amqp1",
      "protocolVersion": "1.0",
      "bindings": {
        "x-ruststream-amqp1": {
          "containerId": "billing-svc"
        }
      }
    }
  },
  "channels": {
    "orders": {
      "bindings": {
        "x-ruststream-amqp1": {
          "address": "orders",
          "capability": "queue",
          "credit": 64,
          "settleMode": "at-least-once"
        }
      }
    }
  }
}

Example​

Example