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

use std::collections::{HashMap,VecDeque};
use generated::basic;
use amq_protocol::types::*;

#[derive(Clone,Debug,PartialEq)]
pub struct Binding {
  pub exchange:    String,
  pub routing_key: String,
  pub no_wait:     bool,
  pub active:      bool,
}

impl Binding {
  pub fn new(exchange: String, routing_key: String, no_wait: bool) -> Binding {
    Binding {
      exchange:    exchange,
      routing_key: routing_key,
      no_wait:     no_wait,
      active:      false,
    }
  }
}

#[derive(Clone,Debug,PartialEq)]
pub struct Message {
  pub delivery_tag: LongLongUInt,
  pub exchange:     String,
  pub routing_key:  String,
  pub redelivered:  bool,
  pub properties:   basic::Properties,
  pub data:         Vec<u8>,
}

impl Message {
  pub fn new(delivery_tag: LongLongUInt, exchange: String, routing_key: String, redelivered: bool) -> Message {
    Message {
      delivery_tag: delivery_tag,
      exchange:     exchange,
      routing_key:  routing_key,
      redelivered:  redelivered,
      properties:   basic::Properties::default(),
      data:         Vec::new(),
    }
  }

  pub fn receive_content(&mut self, data: Vec<u8>) {
    self.data.extend(data);
  }
}

#[derive(Clone,Debug,PartialEq)]
pub struct Consumer {
  pub tag:             String,
  pub no_local:        bool,
  pub no_ack:          bool,
  pub exclusive:       bool,
  pub nowait:          bool,
  pub messages:        VecDeque<Message>,
  pub current_message: Option<Message>,
}

#[derive(Clone,Debug,PartialEq)]
pub struct Queue {
  pub name:                String,
  pub passive:             bool,
  pub durable:             bool,
  pub exclusive:           bool,
  pub auto_delete:         bool,
  pub bindings:            HashMap<(String, String), Binding>,
  pub consumers:           HashMap<String, Consumer>,
  pub message_count:       u32,
  pub consumer_count:      u32,
  pub created:             bool,
  pub get_messages:        VecDeque<Message>,
  pub current_get_message: Option<Message>,
}

impl Queue {
  pub fn new(name: String, passive: bool, durable: bool, exclusive: bool, auto_delete: bool) -> Queue {
    Queue {
      name:                name,
      passive:             passive,
      durable:             durable,
      exclusive:           exclusive,
      auto_delete:         auto_delete,
      bindings:            HashMap::new(),
      consumers:           HashMap::new(),
      message_count:       0,
      consumer_count:      0,
      created:             false,
      get_messages:        VecDeque::new(),
      current_get_message: None,
    }
  }

  pub fn next_message(&mut self, consumer_tag: Option<&str>) -> Option<Message> {
    if let Some(consumer_tag) = consumer_tag {
      self.consumers.get_mut(consumer_tag).and_then(|consumer| consumer.messages.pop_front())
    } else {
      self.get_messages.pop_front()
    }
  }
}