romp 0.5.2

STOMP server and WebSockets platform
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150


use std::sync::{Arc, RwLock};

use log::*;

use chrono::{DateTime, Utc};

use crate::message::stomp_message::{StompMessage, Header};
use crate::session::filter::Filter;
use crate::session::stomp_session::StompSession;
use crate::util::rollover_gt;
use crate::workflow::destination::destination::Destination;
use std::ops::Deref;

#[derive(Debug, PartialEq)]
pub enum Ack {
    Auto,
    Client,
    ClientIndividual,
}

pub fn parse_ack_hdr(hdr: &String) -> Ack {
    match hdr.as_str() {
        "client" => return Ack::Client,
        "client-individual" => return Ack::ClientIndividual,
        _ => Ack::Auto,
    }
}

/// Subscription to a destination, these are stored on the Destination & the session
/// and hold references to the session, so messages can be queued
// TODO could be weak references so we don't have to manually unhook on shutdown

pub struct Subscription {
    session: Arc<RwLock<StompSession>>,
    destination: Arc<RwLock<Destination>>,
    filter: Option<Filter>,
    filter_self: bool,
    id: u64,         // subscription id
    hash_id: String,
    last_msg: usize, // last delivered message
    timestamp: DateTime<Utc>,  // start time
    ack: Ack,        // auto|client|client-individual
}

impl Subscription {
    pub fn new(session: Arc<RwLock<StompSession>>, destination: Arc<RwLock<Destination>>, id: u64, ack: Ack, filter: Option<Filter>, filter_self: bool) -> Subscription {
        let session_id = session.read().unwrap().id();
        debug!("new subscription id={}, ack_mode={:?}, filter_self={}", id, ack, filter_self);
        Subscription {
            session,
            destination,
            filter,
            filter_self,
            id,
            hash_id: format!("{}-{}", session_id, id),
            last_msg: 0,
            timestamp: Utc::now(),
            ack,
        }
    }

    pub fn hash_id(&self) -> &String {
        &self.hash_id
    }

    pub fn user_id(&self) -> String {
        if let Some(user) = &self.session.read().unwrap().user {
            return user.clone();
        } else {
            return String::from("");
        }
    }

    pub fn subscription_id(&self) -> u64 {
        self.id
    }

    pub(crate) fn unsubscribe(&self) -> bool {
        self.destination.write().unwrap().unsubscribe(&self.hash_id)
    }

    /// send a published message to the client, checks first to see if it was already published
    pub(crate) fn publish(&mut self, dest_id: usize, message: Arc<StompMessage>, do_filter: bool) -> bool {
        let msg_id;
        {
            msg_id = message.id;
            if do_filter {
                if let Some(filter) = &self.filter {
                    if !filter.matches_message(message.deref()) {
                        debug!("filtered out {}", msg_id);
                        return false;
                    }
                }
            }
        }

        if self.last_msg == 0 || rollover_gt(msg_id, self.last_msg) {
            let sub_hdr = Header::from_string("subscription", self.id.to_string());
            let ack_hdr = Header::from_string("ack", msg_id.to_string());
            let session = self.session.read().unwrap();
            if self.filter_self {
                if let Some(session_id) = message.session_id {
                    if session_id == session.id() {
                        debug!("not sending to self {}", msg_id);
                        return false;
                    }
                }
            }
            session.send_message_w_hdrs(message.clone(), vec![sub_hdr, ack_hdr]);
            self.last_msg = msg_id;
            debug!("published message id {}", msg_id);
            match self.ack {
                Ack::Auto => {
                    return true;
                },
                _ => {
                    session.pending_ack(msg_id, dest_id);
                    return false;
                },
            }
        } else {
            debug!("already published {}", msg_id);
            return false;
        }

    }

    pub fn as_string(&self) -> String{
        let mut debug = String::new();
        debug.push_str("sub");

        let session = self.session.read().unwrap();
        if let Some(user) = &session.user {
            debug.push_str(" user=\"");
            debug.push_str(user.as_str());
            debug.push_str("\"");
        }
        debug.push_str(" created=");
        debug.push_str(&self.timestamp.to_rfc3339());

        debug
    }
}

impl Drop for Subscription {
    fn drop(&mut self) {
        debug!("dropped");
    }
}