#amqp #rabbitmq #futures #tokio #tls

lapin-futures-tls-internal

Integration of tls engines with lapin-futures

11 unstable releases

✓ Uses Rust 2018 edition

0.7.1 Mar 11, 2020
0.7.0 Mar 3, 2019
0.6.0 Feb 15, 2019
0.3.1 Dec 5, 2018
0.1.3 Nov 14, 2018

#590 in Network programming

Download history 33/week @ 2019-12-01 198/week @ 2019-12-08 178/week @ 2019-12-15 24/week @ 2019-12-22 2/week @ 2019-12-29 36/week @ 2020-01-05 172/week @ 2020-01-12 76/week @ 2020-01-19 54/week @ 2020-01-26 152/week @ 2020-02-02 165/week @ 2020-02-09 26/week @ 2020-02-16 64/week @ 2020-02-23 18/week @ 2020-03-01 56/week @ 2020-03-08 45/week @ 2020-03-15

343 downloads per month
Used in 6 crates (4 directly)

BSD-2-Clause and maybe MPL-2.0

20KB
212 lines

DEPRECATED, use lapin directly instead


lib.rs:

lapin-futures-openssl

This library offers a nice integration of openssl with the lapin-futures library. It uses amq-protocol URI parsing feature and adds the connect and connect_cancellable methods to AMQPUri which will provide you with a lapin_futures::client::Client and optionally a lapin_futures::client::HeartbeatHandle wrapped in a Future.

It autodetects whether you're using amqp or amqps and opens either a raw TcpStream or a TlsStream.

Connecting and opening a channel

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);
        })
    );
}

Dependencies

~14MB
~312K SLoC