solace_rs/session/
builder.rs

1use solace_rs_sys as ffi;
2use std::{
3    ffi::{CString, NulError},
4    marker::PhantomData,
5    mem, ptr,
6};
7
8use crate::{
9    message::InboundMessage,
10    session::SessionEvent,
11    util::{
12        get_last_error_info, on_event_trampoline, on_message_trampoline, static_no_op_on_event,
13        static_no_op_on_message,
14    },
15    Context, Session, SolClientReturnCode, SolClientSubCode,
16};
17
18#[derive(thiserror::Error, Debug)]
19pub enum SessionBuilderError {
20    #[error("session failed to initialize. SolClient return code: {0} subcode: {1}")]
21    InitializationFailure(SolClientReturnCode, SolClientSubCode),
22    #[error("session failed to connect. SolClient return code: {0} subcode: {1}")]
23    ConnectionFailure(SolClientReturnCode, SolClientSubCode),
24    #[error("arg contains interior nul byte")]
25    InvalidArgs(#[from] NulError),
26    #[error("{0} arg need to be set")]
27    MissingRequiredArgs(String),
28    #[error("{0} valid range is {1} foound {2}")]
29    InvalidRange(String, String, String),
30}
31
32type Result<T> = std::result::Result<T, SessionBuilderError>;
33
34fn bool_to_ptr(b: bool) -> *const i8 {
35    if b {
36        ffi::SOLCLIENT_PROP_ENABLE_VAL.as_ptr() as *const i8
37    } else {
38        ffi::SOLCLIENT_PROP_DISABLE_VAL.as_ptr() as *const i8
39    }
40}
41
42struct UncheckedSessionProps<Host, Vpn, Username, Password> {
43    // Note: required params
44    // In the future we can use type state pattern to always force clients to provide these params
45    host_name: Option<Host>,
46    vpn_name: Option<Vpn>,
47    username: Option<Username>,
48    password: Option<Password>,
49
50    // Note: optional params
51    buffer_size_bytes: Option<u64>,
52    block_write_timeout_ms: Option<u64>,
53    connect_timeout_ms: Option<u64>,
54    subconfirm_timeout_ms: Option<u64>,
55    ignore_dup_subscription_error: Option<bool>,
56    tcp_nodelay: Option<bool>,
57    socket_send_buf_size_bytes: Option<u64>,
58    socket_rcv_buf_size_bytes: Option<u64>,
59    keep_alive_interval_ms: Option<u64>,
60    keep_alive_limit: Option<u64>,
61    application_description: Option<Vec<u8>>,
62    client_name: Option<Vec<u8>>,
63    compression_level: Option<u8>,
64    generate_rcv_timestamps: Option<bool>,
65    generate_send_timestamp: Option<bool>,
66    generate_sender_id: Option<bool>,
67    generate_sender_sequence_number: Option<bool>,
68    connect_retries_per_host: Option<i64>,
69    connect_retries: Option<i64>,
70    reconnect_retries: Option<i64>,
71    reconnect_retry_wait_ms: Option<u64>,
72    reapply_subscriptions: Option<bool>,
73    provision_timeout_ms: Option<u64>,
74    calculate_message_expiration: Option<bool>,
75    no_local: Option<bool>,
76    modifyprop_timeout_ms: Option<u64>,
77    ssl_trust_store_dir: Option<Vec<u8>>,
78
79    // TODO: need to check if some of these params will break other assumptions
80    // ex: we might check for ok status on send but if send_blocking is set to false
81    // it will return can_block which will be assumed as an error
82
83    // Note: below params has not exposed
84    // TODO: check if we should even expose this
85    #[allow(dead_code)]
86    send_blocking: Option<bool>,
87    #[allow(dead_code)]
88    subscribe_blocking: Option<bool>,
89    #[allow(dead_code)]
90    block_while_connecting: Option<bool>,
91
92    // TODO: probably should expose this through some other way
93    // maybe a feature flag for the library
94    #[allow(dead_code)]
95    topic_dispatch: Option<bool>,
96}
97
98impl<Host, Vpn, Username, Password> Default
99    for UncheckedSessionProps<Host, Vpn, Username, Password>
100{
101    fn default() -> Self {
102        Self {
103            host_name: None,
104            vpn_name: None,
105            username: None,
106            password: None,
107            buffer_size_bytes: None,
108            block_write_timeout_ms: None,
109            connect_timeout_ms: None,
110            subconfirm_timeout_ms: None,
111            ignore_dup_subscription_error: None,
112            tcp_nodelay: None,
113            socket_send_buf_size_bytes: None,
114            socket_rcv_buf_size_bytes: None,
115            keep_alive_interval_ms: None,
116            keep_alive_limit: None,
117            application_description: None,
118            client_name: None,
119            compression_level: None,
120            generate_rcv_timestamps: None,
121            generate_send_timestamp: None,
122            generate_sender_id: None,
123            generate_sender_sequence_number: None,
124            connect_retries_per_host: None,
125            connect_retries: None,
126            reconnect_retries: None,
127            reconnect_retry_wait_ms: None,
128            reapply_subscriptions: None,
129            provision_timeout_ms: None,
130            calculate_message_expiration: None,
131            no_local: None,
132            modifyprop_timeout_ms: None,
133            send_blocking: None,
134            subscribe_blocking: None,
135            block_while_connecting: None,
136            topic_dispatch: None,
137            ssl_trust_store_dir: None,
138        }
139    }
140}
141
142/// `SessionBuilder` allows setting up a session with customizable options that are not exposed by
143/// the `session` function such as buffer size, timeouts, and more.
144///
145/// For more detailed documentation on all the configuration field, refer to [the official library documentation](https://docs.solace.com/API-Developer-Online-Ref-Documentation/c/group___session_props.html).
146pub struct SessionBuilder<Host, Vpn, Username, Password, OnMessage, OnEvent> {
147    context: Context,
148    props: UncheckedSessionProps<Host, Vpn, Username, Password>,
149
150    // callbacks
151    on_message: Option<OnMessage>,
152    on_event: Option<OnEvent>,
153}
154
155impl<Host, Vpn, Username, Password, OnMessage, OnEvent>
156    SessionBuilder<Host, Vpn, Username, Password, OnMessage, OnEvent>
157{
158    pub(crate) fn new(context: Context) -> Self {
159        Self {
160            context,
161            props: UncheckedSessionProps::default(),
162            on_message: None,
163            on_event: None,
164        }
165    }
166}
167
168impl<'session, Host, Vpn, Username, Password, OnMessage, OnEvent>
169    SessionBuilder<Host, Vpn, Username, Password, OnMessage, OnEvent>
170where
171    Host: Into<Vec<u8>>,
172    Vpn: Into<Vec<u8>>,
173    Username: Into<Vec<u8>>,
174    Password: Into<Vec<u8>>,
175    OnMessage: FnMut(InboundMessage) + Send + 'session,
176    OnEvent: FnMut(SessionEvent) + Send + 'session,
177{
178    pub fn build(mut self) -> Result<Session<'session, OnMessage, OnEvent>> {
179        let config = CheckedSessionProps::try_from(mem::take(&mut self.props))?;
180
181        // Session props is a **char in C
182        // it takes in an array of key and values
183        // first we specify the key, then the value
184        // Session also copies over the props and maintains a copy internally.
185        // Note: Needs to live long enough for the values to be copied
186        let mut session_pt: ffi::solClient_opaqueSession_pt = ptr::null_mut();
187
188        // Box::into_raw(Box::new(Box::new(f))) as *mut _
189        // need to box it twice
190        // first box will result in a fat pointer
191        // causing a seg fault when dereffing in C land.
192        // leaking is also fine since the lifetime of the closure is set to be the lifetime of the
193        // session
194        let (static_on_message_callback, user_on_message, msg_func_ptr) = match self.on_message {
195            Some(f) => {
196                let tramp = on_message_trampoline(&f);
197                let mut func = Box::new(Box::new(f));
198                (tramp, func.as_mut() as *const _ as *mut _, Some(func))
199            }
200            _ => (
201                Some(static_no_op_on_message as unsafe extern "C" fn(_, _, _) -> u32),
202                ptr::null_mut(),
203                None,
204            ),
205        };
206
207        let (static_on_event_callback, user_on_event, event_func_ptr) = match self.on_event {
208            Some(f) => {
209                let tramp = on_event_trampoline(&f);
210                let mut func = Box::new(Box::new(f));
211                (tramp, func.as_mut() as *const _ as *mut _, Some(func))
212            }
213            _ => (
214                Some(static_no_op_on_event as unsafe extern "C" fn(_, _, _)),
215                ptr::null_mut(),
216                None,
217            ),
218        };
219
220        // Function information for Session creation.
221        // The application must set the eventInfo callback information. All Sessions must have an event callback registered.
222        let mut session_func_info: ffi::solClient_session_createFuncInfo_t =
223            ffi::solClient_session_createFuncInfo {
224                rxInfo: ffi::solClient_session_createRxCallbackFuncInfo {
225                    callback_p: ptr::null_mut(),
226                    user_p: ptr::null_mut(),
227                },
228                eventInfo: ffi::solClient_session_createEventCallbackFuncInfo {
229                    callback_p: static_on_event_callback,
230                    user_p: user_on_event,
231                },
232                rxMsgInfo: ffi::solClient_session_createRxMsgCallbackFuncInfo {
233                    callback_p: static_on_message_callback,
234                    user_p: user_on_message,
235                },
236            };
237
238        let mut raw = config.to_raw();
239        let context_ptr = self.context.raw.lock().unwrap();
240        let session_create_raw_rc = unsafe {
241            ffi::solClient_session_create(
242                raw.as_mut_ptr(),
243                context_ptr.ctx,
244                &mut session_pt,
245                &mut session_func_info,
246                std::mem::size_of::<ffi::solClient_session_createFuncInfo_t>(),
247            )
248        };
249        drop(context_ptr);
250
251        let rc = SolClientReturnCode::from_raw(session_create_raw_rc);
252
253        if !rc.is_ok() {
254            let subcode = get_last_error_info();
255            return Err(SessionBuilderError::InitializationFailure(rc, subcode));
256        }
257
258        let connection_raw_rc = unsafe { ffi::solClient_session_connect(session_pt) };
259
260        let rc = SolClientReturnCode::from_raw(connection_raw_rc);
261        if rc.is_ok() {
262            Ok(Session {
263                _msg_fn_ptr: msg_func_ptr,
264                _event_fn_ptr: event_func_ptr,
265                _session_ptr: session_pt,
266                context: self.context,
267                lifetime: PhantomData,
268            })
269        } else {
270            let subcode = get_last_error_info();
271            Err(SessionBuilderError::ConnectionFailure(rc, subcode))
272        }
273    }
274
275    pub fn host_name(mut self, host_name: Host) -> Self {
276        self.props.host_name = Some(host_name);
277        self
278    }
279
280    pub fn vpn_name(mut self, vpn_name: Vpn) -> Self {
281        self.props.vpn_name = Some(vpn_name);
282        self
283    }
284    pub fn username(mut self, username: Username) -> Self {
285        self.props.username = Some(username);
286        self
287    }
288    pub fn password(mut self, password: Password) -> Self {
289        self.props.password = Some(password);
290        self
291    }
292
293    pub fn on_message(mut self, on_message: OnMessage) -> Self {
294        self.on_message = Some(on_message);
295        self
296    }
297
298    pub fn on_event(mut self, on_event: OnEvent) -> Self {
299        self.on_event = Some(on_event);
300        self
301    }
302
303    pub fn buffer_size_bytes(mut self, buffer_size_bytes: u64) -> Self {
304        self.props.buffer_size_bytes = Some(buffer_size_bytes);
305        self
306    }
307    pub fn block_write_timeout_ms(mut self, write_timeout_ms: u64) -> Self {
308        self.props.block_write_timeout_ms = Some(write_timeout_ms);
309        self
310    }
311    pub fn connect_timeout_ms(mut self, connect_timeout_ms: u64) -> Self {
312        self.props.connect_timeout_ms = Some(connect_timeout_ms);
313        self
314    }
315    pub fn subconfirm_timeout_ms(mut self, subconfirm_timeout_ms: u64) -> Self {
316        self.props.subconfirm_timeout_ms = Some(subconfirm_timeout_ms);
317        self
318    }
319    pub fn ignore_dup_subscription_error(mut self, ignore_dup_subscription_error: bool) -> Self {
320        self.props.ignore_dup_subscription_error = Some(ignore_dup_subscription_error);
321        self
322    }
323    pub fn tcp_nodelay(mut self, tcp_nodelay: bool) -> Self {
324        self.props.tcp_nodelay = Some(tcp_nodelay);
325        self
326    }
327    pub fn socket_send_buf_size_bytes(mut self, socket_send_buf_size_bytes: u64) -> Self {
328        self.props.socket_send_buf_size_bytes = Some(socket_send_buf_size_bytes);
329        self
330    }
331    pub fn socket_rcv_buf_size_bytes(mut self, socket_rcv_buf_size_bytes: u64) -> Self {
332        self.props.socket_rcv_buf_size_bytes = Some(socket_rcv_buf_size_bytes);
333        self
334    }
335    pub fn keep_alive_interval_ms(mut self, keep_alive_interval_ms: u64) -> Self {
336        self.props.keep_alive_interval_ms = Some(keep_alive_interval_ms);
337        self
338    }
339    pub fn keep_alive_limit(mut self, keep_alive_limit: u64) -> Self {
340        self.props.keep_alive_limit = Some(keep_alive_limit);
341        self
342    }
343    pub fn application_description<AppDescription: Into<Vec<u8>>>(
344        mut self,
345        application_description: AppDescription,
346    ) -> Self {
347        self.props.application_description = Some(application_description.into());
348        self
349    }
350    pub fn client_name<ClientName: Into<Vec<u8>>>(mut self, client_name: ClientName) -> Self {
351        self.props.client_name = Some(client_name.into());
352        self
353    }
354    pub fn compression_level(mut self, compression_level: u8) -> Self {
355        self.props.compression_level = Some(compression_level);
356        self
357    }
358    pub fn generate_rcv_timestamps(mut self, generate_rcv_timestamps: bool) -> Self {
359        self.props.generate_rcv_timestamps = Some(generate_rcv_timestamps);
360        self
361    }
362    pub fn generate_send_timestamp(mut self, generate_send_timestamp: bool) -> Self {
363        self.props.generate_send_timestamp = Some(generate_send_timestamp);
364        self
365    }
366    pub fn generate_sender_id(mut self, generate_sender_id: bool) -> Self {
367        self.props.generate_sender_id = Some(generate_sender_id);
368        self
369    }
370    pub fn generate_sender_sequence_number(
371        mut self,
372        generate_sender_sequence_number: bool,
373    ) -> Self {
374        self.props.generate_sender_sequence_number = Some(generate_sender_sequence_number);
375        self
376    }
377    pub fn connect_retries_per_host(mut self, connect_retries_per_host: i64) -> Self {
378        self.props.connect_retries_per_host = Some(connect_retries_per_host);
379        self
380    }
381    pub fn connect_retries(mut self, connect_retries: i64) -> Self {
382        self.props.connect_retries = Some(connect_retries);
383        self
384    }
385    pub fn reconnect_retries(mut self, reconnect_retries: i64) -> Self {
386        self.props.reconnect_retries = Some(reconnect_retries);
387        self
388    }
389    pub fn reconnect_retry_wait_ms(mut self, reconnect_retry_wait_ms: u64) -> Self {
390        self.props.reconnect_retry_wait_ms = Some(reconnect_retry_wait_ms);
391        self
392    }
393    pub fn reapply_subscriptions(mut self, reapply_subscriptions: bool) -> Self {
394        self.props.reapply_subscriptions = Some(reapply_subscriptions);
395        self
396    }
397    pub fn provision_timeout_ms(mut self, provision_timeout_ms: u64) -> Self {
398        self.props.provision_timeout_ms = Some(provision_timeout_ms);
399        self
400    }
401    pub fn calculate_message_expiration(mut self, calculate_message_expiration: bool) -> Self {
402        self.props.calculate_message_expiration = Some(calculate_message_expiration);
403        self
404    }
405    pub fn no_local(mut self, no_local: bool) -> Self {
406        self.props.no_local = Some(no_local);
407        self
408    }
409    pub fn modifyprop_timeout_ms(mut self, modifyprop_timeout_ms: u64) -> Self {
410        self.props.modifyprop_timeout_ms = Some(modifyprop_timeout_ms);
411        self
412    }
413    pub fn ssl_trust_store_dir<ClientName: Into<Vec<u8>>>(
414        mut self,
415        ssl_trust_store_dir: ClientName,
416    ) -> Self {
417        self.props.ssl_trust_store_dir = Some(ssl_trust_store_dir.into());
418        self
419    }
420}
421
422struct CheckedSessionProps {
423    host_name: CString,
424    vpn_name: CString,
425    username: CString,
426    password: CString,
427
428    // Note: optional params
429    buffer_size_bytes: Option<CString>,
430    block_write_timeout_ms: Option<CString>,
431    connect_timeout_ms: Option<CString>,
432    subconfirm_timeout_ms: Option<CString>,
433    ignore_dup_subscription_error: Option<bool>,
434    tcp_nodelay: Option<bool>,
435    socket_send_buf_size_bytes: Option<CString>,
436    socket_rcv_buf_size_bytes: Option<CString>,
437    keep_alive_interval_ms: Option<CString>,
438    keep_alive_limit: Option<CString>,
439    application_description: Option<CString>,
440    client_name: Option<CString>,
441    compression_level: Option<CString>,
442    generate_rcv_timestamps: Option<bool>,
443    generate_send_timestamp: Option<bool>,
444    generate_sender_id: Option<bool>,
445    generate_sender_sequence_number: Option<bool>,
446    connect_retries_per_host: Option<CString>,
447    connect_retries: Option<CString>,
448    reconnect_retries: Option<CString>,
449    reconnect_retry_wait_ms: Option<CString>,
450    reapply_subscriptions: Option<bool>,
451    provision_timeout_ms: Option<CString>,
452    calculate_message_expiration: Option<bool>,
453    no_local: Option<bool>,
454    modifyprop_timeout_ms: Option<CString>,
455    ssl_trust_store_dir: Option<CString>,
456}
457
458impl CheckedSessionProps {
459    fn to_raw(&self) -> Vec<*const i8> {
460        let mut props = vec![
461            ffi::SOLCLIENT_SESSION_PROP_HOST.as_ptr() as *const i8,
462            self.host_name.as_ptr(),
463            ffi::SOLCLIENT_SESSION_PROP_VPN_NAME.as_ptr() as *const i8,
464            self.vpn_name.as_ptr(),
465            ffi::SOLCLIENT_SESSION_PROP_USERNAME.as_ptr() as *const i8,
466            self.username.as_ptr(),
467            ffi::SOLCLIENT_SESSION_PROP_PASSWORD.as_ptr() as *const i8,
468            self.password.as_ptr(),
469            ffi::SOLCLIENT_SESSION_PROP_CONNECT_BLOCKING.as_ptr() as *const i8,
470            ffi::SOLCLIENT_PROP_ENABLE_VAL.as_ptr() as *const i8,
471        ];
472
473        if let Some(x) = &self.buffer_size_bytes {
474            props.push(ffi::SOLCLIENT_SESSION_PROP_BUFFER_SIZE.as_ptr() as *const i8);
475            props.push(x.as_ptr());
476        }
477
478        if let Some(x) = &self.block_write_timeout_ms {
479            props.push(ffi::SOLCLIENT_SESSION_PROP_BLOCKING_WRITE_TIMEOUT_MS.as_ptr() as *const i8);
480            props.push(x.as_ptr());
481        }
482        if let Some(x) = &self.connect_timeout_ms {
483            props.push(ffi::SOLCLIENT_SESSION_PROP_CONNECT_TIMEOUT_MS.as_ptr() as *const i8);
484            props.push(x.as_ptr());
485        }
486
487        if let Some(x) = &self.subconfirm_timeout_ms {
488            props.push(ffi::SOLCLIENT_SESSION_PROP_SUBCONFIRM_TIMEOUT_MS.as_ptr() as *const i8);
489            props.push(x.as_ptr());
490        }
491        if let Some(x) = &self.ignore_dup_subscription_error {
492            props.push(
493                ffi::SOLCLIENT_SESSION_PROP_IGNORE_DUP_SUBSCRIPTION_ERROR.as_ptr() as *const i8,
494            );
495            props.push(bool_to_ptr(*x));
496        }
497
498        if let Some(x) = &self.tcp_nodelay {
499            props.push(ffi::SOLCLIENT_SESSION_PROP_TCP_NODELAY.as_ptr() as *const i8);
500            props.push(bool_to_ptr(*x));
501        }
502        if let Some(x) = &self.socket_send_buf_size_bytes {
503            props.push(ffi::SOLCLIENT_SESSION_PROP_SOCKET_SEND_BUF_SIZE.as_ptr() as *const i8);
504            props.push(x.as_ptr());
505        }
506
507        if let Some(x) = &self.socket_rcv_buf_size_bytes {
508            props.push(ffi::SOLCLIENT_SESSION_PROP_SOCKET_RCV_BUF_SIZE.as_ptr() as *const i8);
509            props.push(x.as_ptr());
510        }
511        if let Some(x) = &self.keep_alive_interval_ms {
512            props.push(ffi::SOLCLIENT_SESSION_PROP_KEEP_ALIVE_INT_MS.as_ptr() as *const i8);
513            props.push(x.as_ptr());
514        }
515        if let Some(x) = &self.keep_alive_limit {
516            props.push(ffi::SOLCLIENT_SESSION_PROP_KEEP_ALIVE_LIMIT.as_ptr() as *const i8);
517            props.push(x.as_ptr());
518        }
519        if let Some(x) = &self.application_description {
520            props.push(ffi::SOLCLIENT_SESSION_PROP_APPLICATION_DESCRIPTION.as_ptr() as *const i8);
521            props.push(x.as_ptr());
522        }
523        if let Some(x) = &self.client_name {
524            props.push(ffi::SOLCLIENT_SESSION_PROP_CLIENT_NAME.as_ptr() as *const i8);
525            props.push(x.as_ptr());
526        }
527
528        if let Some(x) = &self.compression_level {
529            props.push(ffi::SOLCLIENT_SESSION_PROP_COMPRESSION_LEVEL.as_ptr() as *const i8);
530            props.push(x.as_ptr());
531        }
532        if let Some(x) = &self.generate_rcv_timestamps {
533            props.push(ffi::SOLCLIENT_SESSION_PROP_GENERATE_RCV_TIMESTAMPS.as_ptr() as *const i8);
534            props.push(bool_to_ptr(*x));
535        }
536        if let Some(x) = &self.generate_send_timestamp {
537            props.push(ffi::SOLCLIENT_SESSION_PROP_GENERATE_SEND_TIMESTAMPS.as_ptr() as *const i8);
538            props.push(bool_to_ptr(*x));
539        }
540        if let Some(x) = &self.generate_sender_id {
541            props.push(ffi::SOLCLIENT_SESSION_PROP_GENERATE_SENDER_ID.as_ptr() as *const i8);
542            props.push(bool_to_ptr(*x));
543        }
544        if let Some(x) = &self.generate_sender_sequence_number {
545            props.push(ffi::SOLCLIENT_SESSION_PROP_GENERATE_SEQUENCE_NUMBER.as_ptr() as *const i8);
546            props.push(bool_to_ptr(*x));
547        }
548        if let Some(x) = &self.connect_retries_per_host {
549            props.push(ffi::SOLCLIENT_SESSION_PROP_CONNECT_RETRIES_PER_HOST.as_ptr() as *const i8);
550            props.push(x.as_ptr());
551        }
552        if let Some(x) = &self.connect_retries {
553            props.push(ffi::SOLCLIENT_SESSION_PROP_CONNECT_RETRIES.as_ptr() as *const i8);
554            props.push(x.as_ptr());
555        }
556        if let Some(x) = &self.reconnect_retries {
557            props.push(ffi::SOLCLIENT_SESSION_PROP_RECONNECT_RETRIES.as_ptr() as *const i8);
558            props.push(x.as_ptr());
559        }
560        if let Some(x) = &self.reconnect_retry_wait_ms {
561            props.push(ffi::SOLCLIENT_SESSION_PROP_RECONNECT_RETRY_WAIT_MS.as_ptr() as *const i8);
562            props.push(x.as_ptr());
563        }
564        if let Some(x) = &self.reapply_subscriptions {
565            props.push(ffi::SOLCLIENT_SESSION_PROP_REAPPLY_SUBSCRIPTIONS.as_ptr() as *const i8);
566            props.push(bool_to_ptr(*x));
567        }
568        if let Some(x) = &self.provision_timeout_ms {
569            props.push(ffi::SOLCLIENT_SESSION_PROP_PROVISION_TIMEOUT_MS.as_ptr() as *const i8);
570            props.push(x.as_ptr());
571        }
572        if let Some(x) = &self.calculate_message_expiration {
573            props.push(
574                ffi::SOLCLIENT_SESSION_PROP_CALCULATE_MESSAGE_EXPIRATION.as_ptr() as *const i8,
575            );
576            props.push(bool_to_ptr(*x));
577        }
578        if let Some(x) = &self.no_local {
579            props.push(ffi::SOLCLIENT_SESSION_PROP_NO_LOCAL.as_ptr() as *const i8);
580            props.push(bool_to_ptr(*x));
581        }
582        if let Some(x) = &self.modifyprop_timeout_ms {
583            props.push(ffi::SOLCLIENT_SESSION_PROP_MODIFYPROP_TIMEOUT_MS.as_ptr() as *const i8);
584            props.push(x.as_ptr());
585        }
586        if let Some(x) = &self.ssl_trust_store_dir {
587            props.push(ffi::SOLCLIENT_SESSION_PROP_SSL_TRUST_STORE_DIR.as_ptr() as *const i8);
588            props.push(x.as_ptr());
589        }
590
591        props.push(ptr::null());
592
593        props
594    }
595}
596
597impl<Host, Vpn, Username, Password> TryFrom<UncheckedSessionProps<Host, Vpn, Username, Password>>
598    for CheckedSessionProps
599where
600    Host: Into<Vec<u8>>,
601    Vpn: Into<Vec<u8>>,
602    Username: Into<Vec<u8>>,
603    Password: Into<Vec<u8>>,
604{
605    type Error = SessionBuilderError;
606
607    fn try_from(
608        value: UncheckedSessionProps<Host, Vpn, Username, Password>,
609    ) -> std::prelude::v1::Result<Self, Self::Error> {
610        let host_name = match value.host_name {
611            Some(x) => CString::new(x)?,
612            None => {
613                return Err(SessionBuilderError::MissingRequiredArgs(
614                    "host_name".to_owned(),
615                ));
616            }
617        };
618
619        let vpn_name = match value.vpn_name {
620            Some(x) => CString::new(x)?,
621            None => {
622                return Err(SessionBuilderError::MissingRequiredArgs(
623                    "vpn_name".to_owned(),
624                ));
625            }
626        };
627
628        let username = match value.username {
629            Some(x) => CString::new(x)?,
630            None => {
631                return Err(SessionBuilderError::MissingRequiredArgs(
632                    "username".to_owned(),
633                ));
634            }
635        };
636
637        let password = match value.password {
638            Some(x) => CString::new(x)?,
639            None => {
640                return Err(SessionBuilderError::MissingRequiredArgs(
641                    "password".to_owned(),
642                ));
643            }
644        };
645
646        let client_name = match value.client_name {
647            Some(x) => Some(CString::new(x)?),
648            None => None,
649        };
650
651        let application_description = match value.application_description {
652            Some(x) => Some(CString::new(x)?),
653            None => None,
654        };
655
656        let buffer_size_bytes = match value.buffer_size_bytes {
657            Some(x) if x < 1 => {
658                return Err(SessionBuilderError::InvalidRange(
659                    "buffer_size_bytes".to_owned(),
660                    ">= 1".to_owned(),
661                    x.to_string(),
662                ));
663            }
664            Some(b) => Some(CString::new(b.to_string())?),
665            None => None,
666        };
667
668        let block_write_timeout_ms = match value.block_write_timeout_ms {
669            Some(x) if x < 1 => {
670                return Err(SessionBuilderError::InvalidRange(
671                    "block_write_timeout_ms".to_owned(),
672                    ">= 1".to_owned(),
673                    x.to_string(),
674                ));
675            }
676            Some(x) => Some(CString::new(x.to_string())?),
677            None => None,
678        };
679
680        let connect_timeout_ms = match value.connect_timeout_ms {
681            Some(x) if x < 1 => {
682                return Err(SessionBuilderError::InvalidRange(
683                    "connect_timeout_ms".to_owned(),
684                    ">= 1".to_owned(),
685                    x.to_string(),
686                ));
687            }
688            Some(x) => Some(CString::new(x.to_string())?),
689            None => None,
690        };
691
692        let subconfirm_timeout_ms = match value.subconfirm_timeout_ms {
693            Some(x) if x < 1000 => {
694                return Err(SessionBuilderError::InvalidRange(
695                    "subconfirm_timeout_ms".to_owned(),
696                    ">= 1000".to_owned(),
697                    x.to_string(),
698                ));
699            }
700            Some(x) => Some(CString::new(x.to_string())?),
701            None => None,
702        };
703
704        let socket_send_buf_size_bytes = match value.socket_send_buf_size_bytes {
705            Some(x) if x != 0 && x < 1024 => {
706                return Err(SessionBuilderError::InvalidRange(
707                    "socket_send_buf_size_bytes".to_owned(),
708                    "0 or >= 1024".to_owned(),
709                    x.to_string(),
710                ));
711            }
712            Some(x) => Some(CString::new(x.to_string())?),
713            None => None,
714        };
715
716        let socket_rcv_buf_size_bytes = match value.socket_rcv_buf_size_bytes {
717            Some(x) if x != 0 && x < 1024 => {
718                return Err(SessionBuilderError::InvalidRange(
719                    "socket_rcv_buf_size_bytes".to_owned(),
720                    "0 or >= 1024".to_owned(),
721                    x.to_string(),
722                ));
723            }
724            Some(x) => Some(CString::new(x.to_string())?),
725            None => None,
726        };
727
728        let keep_alive_interval_ms = match value.keep_alive_interval_ms {
729            Some(x) if x != 0 && x < 50 => {
730                return Err(SessionBuilderError::InvalidRange(
731                    "keep_alive_interval_ms".to_owned(),
732                    "0 or >= 50".to_owned(),
733                    x.to_string(),
734                ));
735            }
736            Some(x) => Some(CString::new(x.to_string())?),
737            None => None,
738        };
739
740        let keep_alive_limit = match value.keep_alive_limit {
741            Some(x) if x < 3 => {
742                return Err(SessionBuilderError::InvalidRange(
743                    "keep_alive_limit".to_owned(),
744                    ">= 3".to_owned(),
745                    x.to_string(),
746                ));
747            }
748            Some(x) => Some(CString::new(x.to_string())?),
749            None => None,
750        };
751
752        let compression_level = match value.compression_level {
753            Some(x) if x > 9 => {
754                return Err(SessionBuilderError::InvalidRange(
755                    "compression_level".to_owned(),
756                    "<= 9".to_owned(),
757                    x.to_string(),
758                ));
759            }
760            Some(x) => Some(CString::new(x.to_string())?),
761            None => None,
762        };
763
764        let connect_retries_per_host = match value.connect_retries_per_host {
765            Some(x) if x < -1 => {
766                return Err(SessionBuilderError::InvalidRange(
767                    "connect_retries_per_host".to_owned(),
768                    ">= -1".to_owned(),
769                    x.to_string(),
770                ));
771            }
772            Some(x) => Some(CString::new(x.to_string())?),
773            None => None,
774        };
775
776        let connect_retries = match value.connect_retries {
777            Some(x) if x < -1 => {
778                return Err(SessionBuilderError::InvalidRange(
779                    "connect_retries ".to_owned(),
780                    ">= -1".to_owned(),
781                    x.to_string(),
782                ));
783            }
784            Some(x) => Some(CString::new(x.to_string())?),
785            None => None,
786        };
787
788        let reconnect_retries = match value.reconnect_retries {
789            Some(x) if x < -1 => {
790                return Err(SessionBuilderError::InvalidRange(
791                    "reconnect_retries ".to_owned(),
792                    ">= -1".to_owned(),
793                    x.to_string(),
794                ));
795            }
796            Some(x) => Some(CString::new(x.to_string())?),
797            None => None,
798        };
799
800        let reconnect_retry_wait_ms = match value.reconnect_retry_wait_ms {
801            Some(x) => Some(CString::new(x.to_string())?),
802            None => None,
803        };
804
805        let provision_timeout_ms = match value.provision_timeout_ms {
806            Some(x) => Some(CString::new(x.to_string())?),
807            None => None,
808        };
809        let modifyprop_timeout_ms = match value.modifyprop_timeout_ms {
810            Some(x) => Some(CString::new(x.to_string())?),
811            None => None,
812        };
813        let ssl_trust_store_dir = match value.ssl_trust_store_dir {
814            Some(x) => Some(CString::new(x)?),
815            None => None,
816        };
817
818        Ok(Self {
819            host_name,
820            vpn_name,
821            username,
822            password,
823            buffer_size_bytes,
824            block_write_timeout_ms,
825            connect_timeout_ms,
826            subconfirm_timeout_ms,
827            ignore_dup_subscription_error: value.ignore_dup_subscription_error,
828            tcp_nodelay: value.tcp_nodelay,
829            socket_send_buf_size_bytes,
830            socket_rcv_buf_size_bytes,
831            keep_alive_interval_ms,
832            keep_alive_limit,
833            application_description,
834            client_name,
835            compression_level,
836            generate_rcv_timestamps: value.generate_rcv_timestamps,
837            generate_send_timestamp: value.generate_send_timestamp,
838            generate_sender_id: value.generate_sender_id,
839            generate_sender_sequence_number: value.generate_sender_sequence_number,
840            connect_retries_per_host,
841            connect_retries,
842            reconnect_retries,
843            reconnect_retry_wait_ms,
844            reapply_subscriptions: value.reapply_subscriptions,
845            provision_timeout_ms,
846            calculate_message_expiration: value.calculate_message_expiration,
847            no_local: value.no_local,
848            modifyprop_timeout_ms,
849            ssl_trust_store_dir,
850        })
851    }
852}
solace_rs/session/builder.rs

solace_rs/session/
builder.rs
