Crate lapin

Crate lapin 

Source
Expand description

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() {
        unsafe { 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

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.
Error
The error that can be returned in this crate.
Queue
RecoveryConfig

Enums§

ChannelState
ConnectionState
ConsumerState
ErrorKind
The type of error that can be returned in this crate.
Event
An event happening on the connection
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