amqp_worker 0.5.9

AMQP Worker to listen and provide trait to process message
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

use amq_protocol_types::{AMQPValue, FieldTable};
use lapin_futures::message::Delivery;

pub fn get_message_death_count(message: &Delivery) -> Option<i64> {
  get_count_from_header(message.properties.headers())
}

fn get_count_from_header(header: &Option<FieldTable>) -> Option<i64> {
  if let Some(header) = header {
    if let Some(death) = header.inner().get("x-death") {
      if let AMQPValue::FieldArray(array) = death {
        let raw_array = array.as_slice();
        if raw_array.is_empty() {
          return None;
        }
        if let AMQPValue::FieldTable(params) = &raw_array[0] {
          if let Some(AMQPValue::LongLongInt(value)) = params.inner().get("count") {
            return Some(*value);
          }
        }
      }
    }
  };
  None
}

#[test]
fn header_information() {
  use std::collections::BTreeMap;

  let header = None;
  let count = get_count_from_header(&header);
  assert!(count == None);

  let mut map = FieldTable::from(BTreeMap::new());
  let mut properties = FieldTable::from(BTreeMap::new());
  properties.insert("count".into(), AMQPValue::LongLongInt(666));

  map.insert("x-death".into(), AMQPValue::FieldArray(vec![].into()));
  let header = Some(map);
  let count = get_count_from_header(&header);
  assert!(count == None);

  let mut map = FieldTable::from(BTreeMap::new());
  let mut properties = FieldTable::from(BTreeMap::new());
  properties.insert("count".into(), AMQPValue::LongLongInt(666));

  map.insert(
    "x-death".into(),
    AMQPValue::FieldArray(vec![AMQPValue::FieldTable(properties).into()].into()),
  );
  let header = Some(map);
  let count = get_count_from_header(&header);
  assert!(count == Some(666));
}