84 releases (32 breaking)

0.33.0 Mar 12, 2020
0.32.5 Mar 9, 2020
0.28.4 Dec 3, 2019
0.28.3 Nov 21, 2019
0.6.0 Mar 30, 2017

#62 in Database interfaces

Download history 447/week @ 2020-06-01 169/week @ 2020-06-08 243/week @ 2020-06-15 94/week @ 2020-06-22 256/week @ 2020-06-29 181/week @ 2020-07-06 208/week @ 2020-07-13 127/week @ 2020-07-20 79/week @ 2020-07-27 13/week @ 2020-08-03 327/week @ 2020-08-10 273/week @ 2020-08-17 175/week @ 2020-08-24 164/week @ 2020-08-31 254/week @ 2020-09-07 122/week @ 2020-09-14

3,607 downloads per month
Used in 13 crates (7 directly)

MIT license

23KB
431 lines

DEPRECATED - use lapin instead


lib.rs:

lapin-futures

This library offers a futures-0.1 based API over the lapin library. It leverages the futures-0.1 library, so you can use it with tokio, futures-cpupool or any other executor.

Publishing a message

use futures::future::Future;
use lapin_futures as lapin;
use crate::lapin::{BasicProperties, Client, ConnectionProperties};
use crate::lapin::options::{BasicPublishOptions, QueueDeclareOptions};
use crate::lapin::types::FieldTable;
use log::info;

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

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

  futures::executor::spawn(
   Client::connect(&addr, ConnectionProperties::default()).and_then(|client| {
      // create_channel returns a future that is resolved
      // once the channel is successfully created
      client.create_channel()
    }).and_then(|channel| {
      let id = channel.id();
      info!("created channel with id: {}", id);

      // we using a "move" closure to reuse the channel
      // once the queue is declared. We could also clone
      // the channel
      channel.queue_declare("hello", QueueDeclareOptions::default(), FieldTable::default()).and_then(move |_| {
        info!("channel {} declared queue {}", id, "hello");

        channel.basic_publish("", "hello", b"hello from tokio".to_vec(), BasicPublishOptions::default(), BasicProperties::default())
      })
    })
  ).wait_future().expect("runtime failure");
}

Creating a consumer

use futures::{Future, Stream};
use lapin_futures as lapin;
use crate::lapin::{Client, ConnectionProperties};
use crate::lapin::options::{BasicConsumeOptions, QueueDeclareOptions};
use crate::lapin::types::FieldTable;
use log::{debug, info};

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

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

  futures::executor::spawn(
   Client::connect(&addr, ConnectionProperties::default()).and_then(|client| {
      // create_channel returns a future that is resolved
      // once the channel is successfully created
      client.create_channel()
    }).and_then(|channel| {
      let id = channel.id();
      info!("created channel with id: {}", id);

      let ch = channel.clone();
      channel.queue_declare("hello", QueueDeclareOptions::default(), FieldTable::default()).and_then(move |queue| {
        info!("channel {} declared queue {:?}", id, queue);

        // basic_consume returns a future of a message
        // stream. Any time a message arrives for this consumer,
        // the for_each method would be called
        channel.basic_consume("hello", "my_consumer", BasicConsumeOptions::default(), FieldTable::default())
      }).and_then(|stream| {
        info!("got consumer stream");

        stream.for_each(move |message| {
          debug!("got message: {:?}", message);
          info!("decoded message: {:?}", std::str::from_utf8(&message.data).unwrap());
          ch.basic_ack(message.delivery_tag, false)
        })
      })
    })
  ).wait_future().expect("runtime failure");
}

Dependencies

~4.5–8.5MB
~194K SLoC