lapin-async 0.22.1

AMQP client library with a low level API designed for use with mio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

use parking_lot::Mutex;

use std::{
  collections::{HashMap, HashSet},
  sync::Arc,
};

use crate::{
  error::{Error, ErrorKind},
  returned_messages::ReturnedMessages,
  wait::{Wait, WaitHandle},
};

#[deprecated(note = "use lapin instead")]
pub type DeliveryTag = u64;

#[derive(Debug, Clone)]
pub(crate) struct Acknowledgements {
  inner: Arc<Mutex<Inner>>,
}

impl Acknowledgements {
  pub(crate) fn new(returned_messages: ReturnedMessages) -> Self {
    Self { inner: Arc::new(Mutex::new(Inner::new(returned_messages))) }
  }

  pub(crate) fn register_pending(&self, delivery_tag: DeliveryTag) {
    self.inner.lock().register_pending(delivery_tag);
  }

  pub(crate) fn get_last_pending(&self) -> Option<Wait<()>> {
    self.inner.lock().last.take()
  }

  pub(crate) fn ack(&self, delivery_tag: DeliveryTag) -> Result<(), Error> {
    self.inner.lock().ack(delivery_tag)
  }

  pub(crate) fn nack(&self, delivery_tag: DeliveryTag) -> Result<(), Error> {
    self.inner.lock().nack(delivery_tag)
  }

  pub(crate) fn ack_all_pending(&self) {
    let mut inner = self.inner.lock();
    for wait in inner.drain_pending() {
      wait.finish(());
    }
  }

  pub(crate) fn nack_all_pending(&self) {
    let mut inner = self.inner.lock();
    for wait in inner.drain_pending() {
      wait.finish(());
    }
  }

  pub(crate) fn ack_all_before(&self, delivery_tag: DeliveryTag) -> Result<(), Error> {
    let mut inner = self.inner.lock();
    for tag in inner.list_pending_before(delivery_tag) {
      inner.ack(tag)?;
    }
    Ok(())
  }

  pub(crate) fn nack_all_before(&self, delivery_tag: DeliveryTag) -> Result<(), Error> {
    let mut inner = self.inner.lock();
    for tag in inner.list_pending_before(delivery_tag) {
      inner.nack(tag)?;
    }
    Ok(())
  }
}

#[derive(Debug)]
struct Inner {
  last:              Option<Wait<()>>,
  pending:           HashMap<DeliveryTag, WaitHandle<()>>,
  returned_messages: ReturnedMessages,
}

impl Inner {
  fn new(returned_messages: ReturnedMessages) -> Self {
    Self {
      last:    None,
      pending: HashMap::default(),
      returned_messages,
    }
  }

  fn register_pending(&mut self, delivery_tag: DeliveryTag) {
    let (wait, wait_handle) = Wait::new();
    self.pending.insert(delivery_tag, wait_handle);
    self.last = Some(wait);
  }

  fn drop_pending(&mut self, delivery_tag: DeliveryTag, success: bool) -> Result<(), Error> {
    if let Some(delivery_wait) =  self.pending.remove(&delivery_tag) {
      if success {
        delivery_wait.finish(());
      } else {
        self.returned_messages.register_waiter(delivery_wait);
      }
      Ok(())
    } else {
      Err(ErrorKind::PreconditionFailed.into())
    }
  }

  fn ack(&mut self, delivery_tag: DeliveryTag) -> Result<(), Error> {
    self.drop_pending(delivery_tag, true)?;
    Ok(())
  }

  fn nack(&mut self, delivery_tag: DeliveryTag) -> Result<(), Error> {
    self.drop_pending(delivery_tag, false)?;
    Ok(())
  }

  fn drain_pending(&mut self) -> Vec<WaitHandle<()>> {
    self.pending.drain().map(|tup| tup.1).collect()
  }

  fn list_pending_before(&mut self, delivery_tag: DeliveryTag) -> HashSet<DeliveryTag> {
    self.pending.iter().map(|tup| tup.0).filter(|tag| **tag <= delivery_tag).cloned().collect()
  }
}