Crate lapin

lapin

This project follows the AMQP 0.9.1 specifications, targeting especially RabbitMQ.

The main access point is the Channel, which contains the individual AMQP methods. As to the AMQP specification, one TCP Connection can contain multiple channels.

Feature switches

• codegen: generate code instead of using pregenerated one
• native-tls: enable amqps support through native-tls (preferred over rustls when set)
• openssl: enable amqps support through openssl (preferred over rustls when set)
• rustls (default): enable amqps support through rustls (uses rustls-native-certs by default)
• rustls-native-certs: same as rustls, be ensure we’ll still use rustls-native-certs even if the default for rustls changes
• rustls-webpki-roots-certs: same as rustls but using webkit-roots instead of rustls-native-certs

Example

use futures_lite::stream::StreamExt;
use lapin::{
    options::*, publisher_confirm::Confirmation, types::FieldTable, BasicProperties, Connection,
    ConnectionProperties, Result,
};
use tracing::info;

fn main() -> Result<()> {
    if std::env::var("RUST_LOG").is_err() {
        std::env::set_var("RUST_LOG", "info");
    }

    tracing_subscriber::fmt::init();

    let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());

    async_global_executor::block_on(async {
        let conn = Connection::connect(
            &addr,
            ConnectionProperties::default(),
        )
        .await?;

        info!("CONNECTED");

        let channel_a = conn.create_channel().await?;
        let channel_b = conn.create_channel().await?;

        let queue = channel_a
            .queue_declare(
                "hello",
                QueueDeclareOptions::default(),
                FieldTable::default(),
            )
            .await?;

        info!(?queue, "Declared queue");

        let mut consumer = channel_b
            .basic_consume(
                "hello",
                "my_consumer",
                BasicConsumeOptions::default(),
                FieldTable::default(),
            )
            .await?;
        async_global_executor::spawn(async move {
            info!("will consume");
            while let Some(delivery) = consumer.next().await {
                let delivery = delivery.expect("error in consumer");
                delivery
                    .ack(BasicAckOptions::default())
                    .await
                    .expect("ack");
            }
        }).detach();

        let payload = b"Hello world!";

        loop {
            let confirm = channel_a
                .basic_publish(
                    "",
                    "hello",
                    BasicPublishOptions::default(),
                    payload,
                    BasicProperties::default(),
                )
                .await?
                .await?;
            assert_eq!(confirm, Confirmation::NotRequested);
        }
    })
}

Re-exports

pub use amq_protocol::tcp;

pub use amq_protocol::types;

pub use amq_protocol::uri;

Modules

acker

auth

Utility to handle SASL authentication with AMQP server

heartbeat

message

options

protocol

The AMQ Protocol implementation (Generated)

publisher_confirm

socket_state

topology

Structs

Channel

Main entry point for most AMQP operations.

ChannelStatus

Configuration

Connection

A TCP connection to the AMQP server.

ConnectionProperties

ConnectionStatus

Consumer

Continuously consumes message from a Queue.

Queue

Enums

ChannelState

ConnectionState

ConsumerState

Error

The type of error that can be returned in this crate.

ExchangeKind

TcpStream

Wrapper around plain or TLS TCP streams

Traits

Connect

Trait providing a method to connect to an AMQP server

ConsumerDelegate

Type Aliases

BasicProperties

Type alias for AMQP BasicProperties

Result

A std Result with a lapin::Error error type