#async-pool #lapin #pool #amqp #async #rabbitmq

yanked cl-deadpool-lapin2

Dead simple async pool for lapin

0.10.0 Feb 23, 2022

#21 in #lapin

MIT/Apache

85KB
1.5K SLoC

Deadpool for Lapin Latest Version Unsafe forbidden Rust 1.54+

Deadpool is a dead simple async pool for connections and objects of any type.

This crate implements a deadpool manager for lapin.

Features

Feature Description Extra dependencies Default
rt_tokio_1 Enable support for tokio crate deadpool/rt_tokio_1 yes
rt_async-std_1 Enable support for async-std crate deadpool/rt_async-std_1 no
serde Enable support for serde crate deadpool/serde, serde/derive no

Example with tokio-amqp crate

use std::sync::Arc;

use deadpool_lapin::{Config, Manager, Pool, Runtime};
use deadpool_lapin::lapin::{
    options::BasicPublishOptions,
    BasicProperties,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut cfg = Config::default();
    cfg.url = Some("amqp://127.0.0.1:5672/%2f".into());
    let pool = cfg.create_pool(Some(Runtime::Tokio1))?;
    for _ in 1..10 {
        let mut connection = pool.get().await?;
        let channel = connection.create_channel().await?;
        channel.basic_publish(
            "",
            "hello",
            BasicPublishOptions::default(),
            b"hello from deadpool",
            BasicProperties::default(),
        ).await?;
    }
    Ok(())
}

Example with config, dotenv and tokio-amqp crate

use std::sync::Arc;

use deadpool_lapin::Runtime;
use deadpool_lapin::lapin::{
    options::BasicPublishOptions,
    BasicProperties,
};
use dotenv::dotenv;
# use serde_1 as serde;

#[derive(Debug, serde::Deserialize)]
# #[serde(crate = "serde_1")]
struct Config {
    #[serde(default)]
    amqp: deadpool_lapin::Config
}

impl Config {
    pub fn from_env() -> Result<Self, config::ConfigError> {
        let mut cfg = config::Config::new();
        cfg.merge(config::Environment::new().separator("__"))?;
        cfg.try_into()
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    dotenv().ok();
    let mut cfg = Config::from_env().unwrap();
    let pool = cfg.amqp.create_pool(Some(Runtime::Tokio1)).unwrap();
    for _ in 1..10 {
        let mut connection = pool.get().await?;
        let channel = connection.create_channel().await?;
        channel.basic_publish(
            "",
            "hello",
            BasicPublishOptions::default(),
            b"hello from deadpool",
            BasicProperties::default(),
        ).await?;
    }
    Ok(())
}

License

Licensed under either of

at your option.

Dependencies

~9–21MB
~325K SLoC