stomp 0.5.1

A full STOMP 1.2 client implementation. Allows programs to interact with message queueing services like ActiveMQ and RabbitMQ.
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

use frame::Frame;
use subscription::AckOrNack::{Ack, Nack};
use std::sync::mpsc::Sender;

#[derive(Copy)]
pub enum AckMode {
  Auto,
  Client,
  ClientIndividual
}

impl AckMode {
  pub fn as_text(&self) -> &'static str {
    match *self {
      AckMode::Auto => "auto",
      AckMode::Client => "client",
      AckMode::ClientIndividual => "client-individual"
    }
  }
}

#[derive(Copy)]
pub enum AckOrNack {
  Ack,
  Nack
}

pub trait MessageHandler {
  fn on_message(&mut self, &Frame) -> AckOrNack;
}

pub struct Subscription <'a> { 
  pub id : String,
  pub topic: String,
  pub ack_mode: AckMode,
  pub handler: Box<MessageHandler + 'a>
}

impl <'a> Subscription <'a> {
  pub fn new(id: u32, topic: &str, ack_mode: AckMode, message_handler: Box<MessageHandler + 'a>) -> Subscription <'a> {
    Subscription {
      id: format!("stomp-rs/{}",id),
      topic: topic.to_string(),
      ack_mode: ack_mode,
      handler: message_handler
    }
  }
}

pub trait ToMessageHandler <'a> {
  fn to_message_handler(self) -> Box<MessageHandler + 'a>;
}

// Support for Sender<T> in subscriptions

struct SendingMessageHandler {
  sender: Sender<Frame>
}

impl MessageHandler for SendingMessageHandler {
  fn on_message(&mut self, frame: &Frame) -> AckOrNack {
    debug!("Sending frame...");
    match self.sender.send(frame.clone()) {
      Ok(_) => Ack,
      Err(error) => {
        error!("Failed to send frame: {}", error);
        Nack
      }
    }
  }
}

impl <'a> ToMessageHandler<'a> for Sender<Frame> {
  fn to_message_handler(self) -> Box<MessageHandler + 'a> {
    Box::new(SendingMessageHandler{sender : self}) as Box<MessageHandler>
  }
}

// Support for closures in subscriptions

struct ClosureMessageHandler <F> where F : FnMut(&Frame) -> AckOrNack {
  closure: F
}

impl <F> MessageHandler for ClosureMessageHandler <F> where F : FnMut(&Frame) -> AckOrNack {
  fn on_message(&mut self, frame: &Frame) -> AckOrNack {
    debug!("Passing frame to closure...");
    (self.closure)(frame)    
  }
}

impl <'a, F> ToMessageHandler<'a> for F where F : FnMut(&Frame) -> AckOrNack + 'a {
  fn to_message_handler(self) -> Box<MessageHandler + 'a> {
    Box::new(ClosureMessageHandler{closure : self}) as Box<MessageHandler>
  }
}