solace_rs/
message.rs

1pub mod destination;
2pub mod inbound;
3pub mod outbound;
4
5use crate::SolClientReturnCode;
6pub use destination::{DestinationType, MessageDestination};
7use enum_primitive::*;
8pub use inbound::InboundMessage;
9pub use outbound::{OutboundMessage, OutboundMessageBuilder};
10use solace_rs_sys as ffi;
11use std::ffi::CStr;
12use std::mem;
13use std::mem::size_of;
14use std::ptr;
15use std::time::{Duration, SystemTime};
16use thiserror::Error;
17
18// the below assertions makes sure that u32 can always be converted into usize safely.
19#[allow(dead_code)]
20const ASSERT_USIZE_IS_AT_LEAST_U32: () = assert!(size_of::<u32>() <= size_of::<usize>());
21
22enum_from_primitive! {
23    #[derive(Debug, PartialEq, Eq)]
24    #[repr(u32)]
25    pub enum DeliveryMode {
26        Direct=ffi::SOLCLIENT_DELIVERY_MODE_DIRECT,
27        Persistent=ffi::SOLCLIENT_DELIVERY_MODE_PERSISTENT,
28        NonPersistent=ffi::SOLCLIENT_DELIVERY_MODE_NONPERSISTENT
29    }
30}
31
32enum_from_primitive! {
33    #[derive(Debug, PartialEq, Eq)]
34    #[repr(u32)]
35    pub enum ClassOfService {
36        One=ffi::SOLCLIENT_COS_1,
37        Two=ffi::SOLCLIENT_COS_2,
38        Three=ffi::SOLCLIENT_COS_3,
39    }
40}
41
42enum_from_primitive! {
43    #[derive(Debug, PartialEq, Eq)]
44    #[repr(i32)]
45    pub enum CacheStatus {
46        InvalidMessage=ffi::solClient_cacheStatus_SOLCLIENT_CACHE_INVALID_MESSAGE,
47        LiveMessage=ffi::solClient_cacheStatus_SOLCLIENT_CACHE_LIVE_MESSAGE,
48        CacheMessage=ffi::solClient_cacheStatus_SOLCLIENT_CACHE_MESSAGE,
49        SuspectMessage=ffi::solClient_cacheStatus_SOLCLIENT_CACHE_SUSPECT_MESSAGE,
50    }
51}
52
53impl From<ClassOfService> for u32 {
54    fn from(val: ClassOfService) -> Self {
55        match val {
56            ClassOfService::One => ffi::SOLCLIENT_COS_1,
57            ClassOfService::Two => ffi::SOLCLIENT_COS_2,
58            ClassOfService::Three => ffi::SOLCLIENT_COS_3,
59        }
60    }
61}
62
63#[derive(Error, Debug)]
64pub enum MessageError {
65    #[error("failed to get field. SolClient return code: {0}")]
66    FieldError(&'static str, SolClientReturnCode),
67    #[error("failed to convert field from solace")]
68    FieldConvertionError(&'static str),
69}
70
71type Result<T> = std::result::Result<T, MessageError>;
72
73pub trait Message<'a> {
74    /// .
75    ///
76    /// # Safety
77    ///
78    /// Should return ptr to a owned valid message.
79    /// No other alias for the ptr should exists.
80    /// Other methods will not check if the message is valid or not
81    ///
82    /// .
83    unsafe fn get_raw_message_ptr(&'a self) -> ffi::solClient_opaqueMsg_pt;
84
85    fn get_payload(&'a self) -> Result<Option<&'a [u8]>> {
86        let mut buffer = ptr::null_mut();
87        let mut buffer_len: u32 = 0;
88
89        let msg_ops_rc = unsafe {
90            ffi::solClient_msg_getBinaryAttachmentPtr(
91                self.get_raw_message_ptr(),
92                &mut buffer,
93                &mut buffer_len,
94            )
95        };
96
97        let rc = SolClientReturnCode::from_raw(msg_ops_rc);
98        match rc {
99            SolClientReturnCode::Ok => (),
100            SolClientReturnCode::NotFound => return Ok(None),
101            _ => return Err(MessageError::FieldError("payload", rc)),
102        }
103
104        // the compile time check ASSERT_USIZE_IS_AT_LEAST_U32 guarantees that this conversion is
105        // possible
106        let buf_len = buffer_len.try_into().unwrap();
107
108        let safe_slice = unsafe { std::slice::from_raw_parts(buffer as *const u8, buf_len) };
109
110        Ok(Some(safe_slice))
111    }
112
113    fn get_application_message_id(&'a self) -> Option<&'a str> {
114        let mut buffer = ptr::null();
115
116        let rc = unsafe {
117            ffi::solClient_msg_getApplicationMessageId(self.get_raw_message_ptr(), &mut buffer)
118        };
119
120        let rc = SolClientReturnCode::from_raw(rc);
121
122        if !rc.is_ok() {
123            return None;
124        }
125
126        let c_str = unsafe { CStr::from_ptr(buffer) };
127
128        c_str.to_str().ok()
129    }
130
131    fn get_application_msg_type(&'a self) -> Option<&'a str> {
132        let mut buffer = ptr::null();
133
134        let rc = unsafe {
135            ffi::solClient_msg_getApplicationMsgType(self.get_raw_message_ptr(), &mut buffer)
136        };
137
138        let rc = SolClientReturnCode::from_raw(rc);
139
140        if !rc.is_ok() {
141            return None;
142        }
143
144        let c_str = unsafe { CStr::from_ptr(buffer) };
145
146        c_str.to_str().ok()
147    }
148
149    fn get_class_of_service(&'a self) -> Result<ClassOfService> {
150        let mut cos: u32 = 0;
151        let rc =
152            unsafe { ffi::solClient_msg_getClassOfService(self.get_raw_message_ptr(), &mut cos) };
153
154        let rc = SolClientReturnCode::from_raw(rc);
155        if !rc.is_ok() {
156            return Err(MessageError::FieldError("ClassOfService", rc));
157        }
158
159        let Some(cos) = ClassOfService::from_u32(cos) else {
160            return Err(MessageError::FieldConvertionError("ClassOfService"));
161        };
162
163        Ok(cos)
164    }
165
166    fn get_correlation_id(&'a self) -> Result<Option<&'a str>> {
167        let mut buffer = ptr::null();
168
169        let rc =
170            unsafe { ffi::solClient_msg_getCorrelationId(self.get_raw_message_ptr(), &mut buffer) };
171
172        let rc = SolClientReturnCode::from_raw(rc);
173        match rc {
174            SolClientReturnCode::Ok => (),
175            SolClientReturnCode::NotFound => return Ok(None),
176            _ => return Err(MessageError::FieldError("correlation_id", rc)),
177        }
178
179        let c_str = unsafe { CStr::from_ptr(buffer) };
180
181        let str = c_str
182            .to_str()
183            .map_err(|_| MessageError::FieldConvertionError("correlation_id"))?;
184
185        Ok(Some(str))
186    }
187
188    fn is_eliding_eligible(&'a self) -> bool {
189        let unsafe_result =
190            unsafe { ffi::solClient_msg_isElidingEligible(self.get_raw_message_ptr()) };
191
192        unsafe_result != 0
193    }
194
195    fn get_expiration(&'a self) -> i64 {
196        let mut exp: i64 = 0;
197        unsafe { ffi::solClient_msg_getExpiration(self.get_raw_message_ptr(), &mut exp) };
198
199        exp
200    }
201
202    fn get_priority(&'a self) -> Result<Option<u8>> {
203        let mut priority: i32 = 0;
204        let rc =
205            unsafe { ffi::solClient_msg_getPriority(self.get_raw_message_ptr(), &mut priority) };
206
207        let rc = SolClientReturnCode::from_raw(rc);
208        if !rc.is_ok() {
209            return Err(MessageError::FieldError("priority", rc));
210        }
211
212        if priority == -1 {
213            return Ok(None);
214        }
215
216        Ok(Some(priority as u8))
217    }
218
219    fn get_sequence_number(&'a self) -> Result<Option<i64>> {
220        let mut seq_num: i64 = 0;
221        let rc = unsafe {
222            ffi::solClient_msg_getSequenceNumber(self.get_raw_message_ptr(), &mut seq_num)
223        };
224        let rc = SolClientReturnCode::from_raw(rc);
225
226        match rc {
227            SolClientReturnCode::Ok => Ok(Some(seq_num)),
228            SolClientReturnCode::NotFound => Ok(None),
229            _ => Err(MessageError::FieldError("sequence_number", rc)),
230        }
231    }
232
233    fn get_destination(&'a self) -> Result<Option<MessageDestination>> {
234        let mut dest_struct: ffi::solClient_destination = ffi::solClient_destination {
235            destType: ffi::solClient_destinationType_SOLCLIENT_NULL_DESTINATION,
236            dest: ptr::null_mut(),
237        };
238
239        let rc = unsafe {
240            ffi::solClient_msg_getDestination(
241                self.get_raw_message_ptr(),
242                &mut dest_struct,
243                mem::size_of::<ffi::solClient_destination>(),
244            )
245        };
246
247        let rc = SolClientReturnCode::from_raw(rc);
248
249        match rc {
250            SolClientReturnCode::NotFound => Ok(None),
251            SolClientReturnCode::Fail => Err(MessageError::FieldError("destination", rc)),
252            _ => Ok(Some(MessageDestination::from(dest_struct))),
253        }
254    }
255
256    fn get_reply_to(&'a self) -> Result<Option<MessageDestination>> {
257        let mut dest_struct: ffi::solClient_destination = ffi::solClient_destination {
258            destType: ffi::solClient_destinationType_SOLCLIENT_NULL_DESTINATION,
259            dest: ptr::null_mut(),
260        };
261
262        let rc = unsafe {
263            ffi::solClient_msg_getReplyTo(
264                self.get_raw_message_ptr(),
265                &mut dest_struct,
266                mem::size_of::<ffi::solClient_destination>(),
267            )
268        };
269
270        let rc = SolClientReturnCode::from_raw(rc);
271
272        match rc {
273            SolClientReturnCode::NotFound => Ok(None),
274            SolClientReturnCode::Fail => Err(MessageError::FieldError("destination", rc)),
275            _ => Ok(Some(MessageDestination::from(dest_struct))),
276        }
277    }
278
279    fn is_reply(&'a self) -> bool {
280        let res = unsafe { ffi::solClient_msg_isReplyMsg(self.get_raw_message_ptr()) };
281        res != 0
282    }
283
284    fn get_sender_timestamp(&'a self) -> Result<Option<SystemTime>> {
285        let mut ts: i64 = 0;
286        let rc =
287            unsafe { ffi::solClient_msg_getSenderTimestamp(self.get_raw_message_ptr(), &mut ts) };
288
289        let rc = SolClientReturnCode::from_raw(rc);
290
291        match rc {
292            SolClientReturnCode::NotFound => Ok(None),
293            SolClientReturnCode::Ok => Ok(Some(
294                SystemTime::UNIX_EPOCH + Duration::from_millis(ts.try_into().unwrap()),
295            )),
296            _ => Err(MessageError::FieldError("sender_timestamp", rc)),
297        }
298    }
299
300    fn get_user_data(&'a self) -> Result<Option<&'a [u8]>> {
301        let mut buffer = ptr::null_mut();
302        let mut buffer_len: u32 = 0;
303
304        let rc = unsafe {
305            ffi::solClient_msg_getUserDataPtr(
306                self.get_raw_message_ptr(),
307                &mut buffer,
308                &mut buffer_len,
309            )
310        };
311
312        let rc = SolClientReturnCode::from_raw(rc);
313        match rc {
314            SolClientReturnCode::Ok => (),
315            SolClientReturnCode::NotFound => return Ok(None),
316            _ => return Err(MessageError::FieldError("user_data", rc)),
317        }
318
319        // the compile time check ASSERT_USIZE_IS_AT_LEAST_U32 guarantees that this conversion is
320        // possible
321        let buf_len = buffer_len.try_into().unwrap();
322
323        let safe_slice = unsafe { std::slice::from_raw_parts(buffer as *const u8, buf_len) };
324
325        Ok(Some(safe_slice))
326    }
327}
solace_rs/message.rs

solace_rs/
message.rs
