#future #tls #amqp #tokio #rabbitmq #api-bindings

lapin-futures-tls-internal

lapin-futures与tls引擎的集成

11个不稳定版本

0.7.1 2020年3月11日
0.7.0 2019年3月3日
0.6.0 2019年2月15日
0.3.1 2018年12月5日
0.1.3 2018年11月14日

#56#rabbitmq

每月40次下载
6 个crate中使用 (直接使用4个)

BSD-2-Clause

22KB
211

已弃用,直接使用lapin


lib.rs:

lapin-futures-openssl

此库为openssl与lapin-futures库提供了一个良好的集成。它使用amq-protocol URI解析功能,并在AMQPUri上添加了connectconnect_cancellable方法,这将为您提供包装在Future中的lapin_futures::client::Client和可选的lapin_futures::client::HeartbeatHandle

它自动检测您是否使用amqpamqps,并打开一个原始的TcpStream或一个TlsStream

连接和打开通道

use env_logger;
use failure::Error;
use futures::{self, future::Future};
use lapin_futures_tls_internal::{AMQPConnectionTlsExt, lapin};
use lapin::channel::ConfirmSelectOptions;
use native_tls;
use tokio;
use tokio_tls::TlsConnector;

use std::io;

fn main() {
    env_logger::init();

    tokio::run(
        "amqps://user:pass@host/vhost?heartbeat=10".connect_cancellable(|err| {
            eprintln!("heartbeat error: {:?}", err);
        }, |host, stream| {
            Box::new(futures::future::result(native_tls::TlsConnector::builder().build().map_err(|_| io::Error::new(io::ErrorKind::Other, "Failed to create connector"))).and_then(move |connector| {
                TlsConnector::from(connector).connect(&host, stream).map_err(|_| io::Error::new(io::ErrorKind::Other, "Failed to connect")).map(Box::new)
            }))
        }).map_err(Error::from).and_then(|(client, heartbeat_handle)| {
            println!("Connected!");
            client.create_confirm_channel(ConfirmSelectOptions::default()).map(|channel| (channel, heartbeat_handle)).and_then(|(channel, heartbeat_handle)| {
                println!("Stopping heartbeat.");
                heartbeat_handle.stop();
                println!("Closing channel.");
                channel.close(200, "Bye")
            }).map_err(Error::from)
        }).map_err(|err| {
            eprintln!("amqp error: {:?}", err);
        })
    );
}

依赖项

~17–31MB
~510K SLoC