clickhouse_rs_async/types/
options.rs

1use std::{
2    borrow::Cow,
3    collections::HashMap,
4    fmt,
5    str::FromStr,
6    sync::{Arc, Mutex},
7    time::Duration,
8};
9
10use crate::errors::{Error, Result, UrlError};
11use percent_encoding::percent_decode;
12use url::Url;
13
14const DEFAULT_MIN_CONNS: usize = 10;
15
16const DEFAULT_MAX_CONNS: usize = 20;
17
18#[derive(Debug)]
19#[allow(clippy::large_enum_variant)]
20enum State {
21    Raw(Options),
22    Url(String),
23}
24
25#[derive(Clone)]
26pub struct OptionsSource {
27    state: Arc<Mutex<State>>,
28}
29
30impl fmt::Debug for OptionsSource {
31    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
32        let guard = self.state.lock().unwrap();
33        match *guard {
34            State::Url(ref url) => write!(f, "Url({url})"),
35            State::Raw(ref options) => write!(f, "{options:?}"),
36        }
37    }
38}
39
40impl OptionsSource {
41    pub(crate) fn get(&self) -> Result<Cow<Options>> {
42        let mut state = self.state.lock().unwrap();
43        loop {
44            *state = match &*state {
45                State::Raw(ref options) => {
46                    let ptr = options as *const Options;
47                    return unsafe { Ok(Cow::Borrowed(ptr.as_ref().unwrap())) };
48                }
49                State::Url(url) => {
50                    let options = from_url(url)?;
51                    State::Raw(options)
52                }
53            };
54        }
55    }
56}
57
58impl Default for OptionsSource {
59    fn default() -> Self {
60        Self {
61            state: Arc::new(Mutex::new(State::Raw(Options::default()))),
62        }
63    }
64}
65
66pub trait IntoOptions {
67    fn into_options_src(self) -> OptionsSource;
68}
69
70impl IntoOptions for Options {
71    fn into_options_src(self) -> OptionsSource {
72        OptionsSource {
73            state: Arc::new(Mutex::new(State::Raw(self))),
74        }
75    }
76}
77
78impl IntoOptions for &str {
79    fn into_options_src(self) -> OptionsSource {
80        OptionsSource {
81            state: Arc::new(Mutex::new(State::Url(self.into()))),
82        }
83    }
84}
85
86impl IntoOptions for String {
87    fn into_options_src(self) -> OptionsSource {
88        OptionsSource {
89            state: Arc::new(Mutex::new(State::Url(self))),
90        }
91    }
92}
93
94/// An X509 certificate for native-tls.
95#[cfg(feature = "tls-native-tls")]
96#[derive(Clone)]
97pub struct Certificate(Arc<native_tls::Certificate>);
98#[cfg(feature = "tls-native-tls")]
99impl Certificate {
100    /// Parses a DER-formatted X509 certificate.
101    pub fn from_der(der: &[u8]) -> Result<Certificate> {
102        let inner = match native_tls::Certificate::from_der(der) {
103            Ok(certificate) => certificate,
104            Err(err) => return Err(Error::Other(err.to_string().into())),
105        };
106        Ok(Certificate(Arc::new(inner)))
107    }
108
109    /// Parses a PEM-formatted X509 certificate.
110    pub fn from_pem(der: &[u8]) -> Result<Certificate> {
111        let inner = match native_tls::Certificate::from_pem(der) {
112            Ok(certificate) => certificate,
113            Err(err) => return Err(Error::Other(err.to_string().into())),
114        };
115        Ok(Certificate(Arc::new(inner)))
116    }
117}
118#[cfg(feature = "tls-native-tls")]
119impl From<Certificate> for native_tls::Certificate {
120    fn from(value: Certificate) -> Self {
121        value.0.as_ref().clone()
122    }
123}
124
125/// An X509 certificate for rustls.
126#[cfg(feature = "tls-rustls")]
127#[derive(Clone)]
128pub struct Certificate(Arc<Vec<rustls::pki_types::CertificateDer<'static>>>);
129#[cfg(feature = "tls-rustls")]
130impl Certificate {
131    /// Parses a DER-formatted X509 certificate.
132    pub fn from_der(der: &[u8]) -> Result<Certificate> {
133        let der = der.to_vec();
134        let inner = match rustls::pki_types::CertificateDer::try_from(der) {
135            Ok(certificate) => certificate,
136            Err(err) => return Err(Error::Other(err.to_string().into())),
137        };
138        Ok(Certificate(Arc::new(vec![inner])))
139    }
140
141    /// Parses a PEM-formatted X509 certificate.
142    pub fn from_pem(der: &[u8]) -> Result<Certificate> {
143        let certs = rustls_pemfile::certs(&mut der.as_ref())
144            .map(|result| result.unwrap())
145            .collect();
146        Ok(Certificate(Arc::new(certs)))
147    }
148}
149#[cfg(feature = "tls-rustls")]
150impl From<Certificate> for Vec<rustls::pki_types::CertificateDer<'static>> {
151    fn from(value: Certificate) -> Self {
152        value.0.as_ref().clone()
153    }
154}
155
156#[cfg(feature = "_tls")]
157impl fmt::Debug for Certificate {
158    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
159        write!(f, "[Certificate]")
160    }
161}
162#[cfg(feature = "_tls")]
163impl PartialEq for Certificate {
164    fn eq(&self, _other: &Self) -> bool {
165        true
166    }
167}
168
169#[derive(Clone, PartialEq, Debug)]
170pub enum SettingType {
171    String(String),
172    Bool(bool),
173    UInt64(u64),
174    Float64(f64),
175}
176
177impl fmt::Display for SettingType {
178    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
179        match &self {
180            SettingType::Bool(val) => write!(f, "{val}"),
181            SettingType::UInt64(val) => write!(f, "{val}"),
182            SettingType::Float64(val) => write!(f, "{val}"),
183            SettingType::String(val) => write!(f, "{val}"),
184        }
185    }
186}
187
188impl From<&str> for SettingType {
189    fn from(val: &str) -> Self {
190        SettingType::String(val.into())
191    }
192}
193
194impl From<bool> for SettingType {
195    fn from(val: bool) -> Self {
196        SettingType::Bool(val)
197    }
198}
199
200impl From<u64> for SettingType {
201    fn from(val: u64) -> Self {
202        SettingType::UInt64(val)
203    }
204}
205
206impl From<i32> for SettingType {
207    fn from(val: i32) -> Self {
208        SettingType::UInt64(val as u64)
209    }
210}
211
212impl From<i64> for SettingType {
213    fn from(val: i64) -> Self {
214        SettingType::UInt64(val as u64)
215    }
216}
217
218impl From<f64> for SettingType {
219    fn from(val: f64) -> Self {
220        SettingType::Float64(val)
221    }
222}
223
224#[derive(Clone, PartialEq, Debug)]
225pub struct SettingValue {
226    pub(crate) value: SettingType,
227    pub(crate) is_important: bool,
228}
229impl fmt::Display for SettingValue {
230    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
231        self.value.fmt(f)
232    }
233}
234
235/// Clickhouse connection options.
236#[derive(Clone, PartialEq)]
237pub struct Options {
238    /// Address of clickhouse server (defaults to `127.0.0.1:9000`).
239    pub(crate) addr: Url,
240
241    /// Database name. (defaults to `default`).
242    pub(crate) database: String,
243    /// User name (defaults to `default`).
244    pub(crate) username: String,
245    /// Access password (defaults to `""`).
246    pub(crate) password: String,
247
248    /// Enable compression (defaults to `false`).
249    pub(crate) compression: bool,
250
251    /// Lower bound of opened connections for `Pool` (defaults to 10).
252    pub(crate) pool_min: usize,
253    /// Upper bound of opened connections for `Pool` (defaults to 20).
254    pub(crate) pool_max: usize,
255
256    /// Whether to enable `TCP_NODELAY` (defaults to `true`).
257    pub(crate) nodelay: bool,
258    /// TCP keep alive timeout in milliseconds (defaults to `None`).
259    pub(crate) keepalive: Option<Duration>,
260
261    /// Ping server every time before execute any query. (defaults to `true`)
262    pub(crate) ping_before_query: bool,
263    /// Count of retry to send request to server. (defaults to `3`)
264    pub(crate) send_retries: usize,
265    /// Amount of time to wait before next retry. (defaults to `5 sec`)
266    pub(crate) retry_timeout: Duration,
267    /// Timeout for ping (defaults to `500 ms`)
268    pub(crate) ping_timeout: Duration,
269
270    /// Timeout for connection (defaults to `500 ms`)
271    pub(crate) connection_timeout: Duration,
272
273    /// Timeout for queries (defaults to `180 sec`)
274    pub(crate) query_timeout: Duration,
275
276    /// Timeout for inserts (defaults to `180 sec`)
277    pub(crate) insert_timeout: Option<Duration>,
278
279    /// Timeout for execute (defaults to `180 sec`)
280    pub(crate) execute_timeout: Option<Duration>,
281
282    /// Enable TLS encryption (defaults to `false`)
283    #[cfg(feature = "_tls")]
284    pub(crate) secure: bool,
285
286    /// Skip certificate verification (default is `false`).
287    #[cfg(feature = "_tls")]
288    pub(crate) skip_verify: bool,
289
290    /// An X509 certificate.
291    #[cfg(feature = "_tls")]
292    pub(crate) certificate: Option<Certificate>,
293
294    /// Query settings
295    pub(crate) settings: HashMap<String, SettingValue>,
296
297    /// Comma separated list of single address host for load-balancing.
298    pub(crate) alt_hosts: Vec<Url>,
299}
300
301impl fmt::Debug for Options {
302    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
303        f.debug_struct("Options")
304            .field("addr", &self.addr)
305            .field("database", &self.database)
306            .field("compression", &self.compression)
307            .field("pool_min", &self.pool_min)
308            .field("pool_max", &self.pool_max)
309            .field("nodelay", &self.nodelay)
310            .field("keepalive", &self.keepalive)
311            .field("ping_before_query", &self.ping_before_query)
312            .field("send_retries", &self.send_retries)
313            .field("retry_timeout", &self.retry_timeout)
314            .field("ping_timeout", &self.ping_timeout)
315            .field("connection_timeout", &self.connection_timeout)
316            .field("settings", &self.settings)
317            .field("alt_hosts", &self.alt_hosts)
318            .finish()
319    }
320}
321
322impl Default for Options {
323    fn default() -> Self {
324        Self {
325            addr: Url::parse("tcp://default@127.0.0.1:9000").unwrap(),
326            database: "default".into(),
327            username: "default".into(),
328            password: "".into(),
329            compression: false,
330            pool_min: DEFAULT_MIN_CONNS,
331            pool_max: DEFAULT_MAX_CONNS,
332            nodelay: true,
333            keepalive: None,
334            ping_before_query: true,
335            send_retries: 3,
336            retry_timeout: Duration::from_secs(5),
337            ping_timeout: Duration::from_millis(500),
338            connection_timeout: Duration::from_millis(500),
339            query_timeout: Duration::from_secs(180),
340            insert_timeout: Some(Duration::from_secs(180)),
341            execute_timeout: Some(Duration::from_secs(180)),
342            #[cfg(feature = "_tls")]
343            secure: false,
344            #[cfg(feature = "_tls")]
345            skip_verify: false,
346            #[cfg(feature = "_tls")]
347            certificate: None,
348            settings: HashMap::new(),
349            alt_hosts: Vec::new(),
350        }
351    }
352}
353
354macro_rules! property {
355    ( $k:ident: $t:ty ) => {
356        pub fn $k(self, $k: $t) -> Self {
357            Self {
358                $k: $k.into(),
359                ..self
360            }
361        }
362    };
363    ( $(#[$attr:meta])* => $k:ident: $t:ty ) => {
364        $(#[$attr])*
365        pub fn $k(self, $k: $t) -> Self {
366            Self {
367                $k: $k.into(),
368                ..self
369            }
370        }
371    }
372}
373
374impl Options {
375    /// Constructs a new Options.
376    pub fn new<A>(addr: A) -> Self
377    where
378        A: Into<Url>,
379    {
380        Self {
381            addr: addr.into(),
382            ..Self::default()
383        }
384    }
385
386    pub fn with_setting<V>(mut self, name: &str, value: V, is_important: bool) -> Self
387    where
388        V: Into<SettingType>,
389    {
390        let value: SettingType = value.into();
391        self.settings.insert(
392            name.into(),
393            SettingValue {
394                value,
395                is_important,
396            },
397        );
398        self
399    }
400
401    property! {
402        /// Database name. (defaults to `default`).
403        => database: &str
404    }
405
406    property! {
407        /// User name (defaults to `default`).
408        => username: &str
409    }
410
411    property! {
412        /// Access password (defaults to `""`).
413        => password: &str
414    }
415
416    /// Enable compression (defaults to `false`).
417    pub fn with_compression(self) -> Self {
418        Self {
419            compression: true,
420            ..self
421        }
422    }
423
424    property! {
425        /// Lower bound of opened connections for `Pool` (defaults to `10`).
426        => pool_min: usize
427    }
428
429    property! {
430        /// Upper bound of opened connections for `Pool` (defaults to `20`).
431        => pool_max: usize
432    }
433
434    property! {
435        /// Whether to enable `TCP_NODELAY` (defaults to `true`).
436        => nodelay: bool
437    }
438
439    property! {
440        /// TCP keep alive timeout in milliseconds (defaults to `None`).
441        => keepalive: Option<Duration>
442    }
443
444    property! {
445        /// Ping server every time before execute any query. (defaults to `true`).
446        => ping_before_query: bool
447    }
448
449    property! {
450        /// Count of retry to send request to server. (defaults to `3`).
451        => send_retries: usize
452    }
453
454    property! {
455        /// Amount of time to wait before next retry. (defaults to `5 sec`).
456        => retry_timeout: Duration
457    }
458
459    property! {
460        /// Timeout for ping (defaults to `500 ms`).
461        => ping_timeout: Duration
462    }
463
464    property! {
465        /// Timeout for connection (defaults to `500 ms`).
466        => connection_timeout: Duration
467    }
468
469    property! {
470        /// Timeout for query (defaults to `180,000 ms`).
471        => query_timeout: Duration
472    }
473
474    property! {
475        /// Timeout for insert (defaults to `180,000 ms`).
476        => insert_timeout: Option<Duration>
477    }
478
479    property! {
480        /// Timeout for execute (defaults to `180 sec`).
481        => execute_timeout: Option<Duration>
482    }
483
484    #[cfg(feature = "_tls")]
485    property! {
486        /// Establish secure connection (default is `false`).
487        => secure: bool
488    }
489
490    #[cfg(feature = "_tls")]
491    property! {
492        /// Skip certificate verification (default is `false`).
493        => skip_verify: bool
494    }
495
496    #[cfg(feature = "_tls")]
497    property! {
498        /// An X509 certificate.
499        => certificate: Option<Certificate>
500    }
501
502    property! {
503        /// Query settings
504        => settings: HashMap<String, SettingValue>
505    }
506
507    property! {
508        /// Comma separated list of single address host for load-balancing.
509        => alt_hosts: Vec<Url>
510    }
511}
512
513impl FromStr for Options {
514    type Err = Error;
515
516    fn from_str(url: &str) -> Result<Self> {
517        from_url(url)
518    }
519}
520
521fn from_url(url_str: &str) -> Result<Options> {
522    let url = Url::parse(url_str)?;
523
524    if url.scheme() != "tcp" {
525        return Err(UrlError::UnsupportedScheme {
526            scheme: url.scheme().to_string(),
527        }
528        .into());
529    }
530
531    if url.cannot_be_a_base() || !url.has_host() {
532        return Err(UrlError::Invalid.into());
533    }
534
535    let mut options = Options::default();
536
537    if let Some(username) = get_username_from_url(&url) {
538        options.username = username.into();
539    }
540
541    if let Some(password) = get_password_from_url(&url) {
542        options.password = password.into()
543    }
544
545    let mut addr = url.clone();
546    addr.set_path("");
547    addr.set_query(None);
548
549    let port = url.port().or(Some(9000));
550    addr.set_port(port).map_err(|_| UrlError::Invalid)?;
551    options.addr = addr;
552
553    if let Some(database) = get_database_from_url(&url)? {
554        options.database = database.into();
555    }
556
557    set_params(&mut options, url.query_pairs())?;
558
559    Ok(options)
560}
561
562fn set_params<'a, I>(options: &mut Options, iter: I) -> std::result::Result<(), UrlError>
563where
564    I: Iterator<Item = (Cow<'a, str>, Cow<'a, str>)>,
565{
566    for (key, value) in iter {
567        match key.as_ref() {
568            "pool_min" => options.pool_min = parse_param(key, value, usize::from_str)?,
569            "pool_max" => options.pool_max = parse_param(key, value, usize::from_str)?,
570            "nodelay" => options.nodelay = parse_param(key, value, bool::from_str)?,
571            "keepalive" => options.keepalive = parse_param(key, value, parse_opt_duration)?,
572            "ping_before_query" => {
573                options.ping_before_query = parse_param(key, value, bool::from_str)?
574            }
575            "send_retries" => options.send_retries = parse_param(key, value, usize::from_str)?,
576            "retry_timeout" => options.retry_timeout = parse_param(key, value, parse_duration)?,
577            "ping_timeout" => options.ping_timeout = parse_param(key, value, parse_duration)?,
578            "connection_timeout" => {
579                options.connection_timeout = parse_param(key, value, parse_duration)?
580            }
581            "query_timeout" => options.query_timeout = parse_param(key, value, parse_duration)?,
582            "insert_timeout" => {
583                options.insert_timeout = parse_param(key, value, parse_opt_duration)?
584            }
585            "execute_timeout" => {
586                options.execute_timeout = parse_param(key, value, parse_opt_duration)?
587            }
588            "compression" => options.compression = parse_param(key, value, parse_compression)?,
589            #[cfg(feature = "_tls")]
590            "secure" => options.secure = parse_param(key, value, bool::from_str)?,
591            #[cfg(feature = "_tls")]
592            "skip_verify" => options.skip_verify = parse_param(key, value, bool::from_str)?,
593            "alt_hosts" => options.alt_hosts = parse_param(key, value, parse_hosts)?,
594            _ => {
595                let value = SettingType::String(value.to_string());
596                options.settings.insert(
597                    key.to_string(),
598                    SettingValue {
599                        value,
600                        is_important: true,
601                    },
602                );
603            }
604        };
605    }
606
607    Ok(())
608}
609
610fn parse_param<'a, F, T, E>(
611    param: Cow<'a, str>,
612    value: Cow<'a, str>,
613    parse: F,
614) -> std::result::Result<T, UrlError>
615where
616    F: Fn(&str) -> std::result::Result<T, E>,
617{
618    let source = percent_decode(value.as_bytes()).decode_utf8_lossy();
619    match parse(source.as_ref()) {
620        Ok(value) => Ok(value),
621        Err(_) => Err(UrlError::InvalidParamValue {
622            param: param.into(),
623            value: value.into(),
624        }),
625    }
626}
627
628fn get_username_from_url(url: &Url) -> Option<Cow<'_, str>> {
629    let user = url.username();
630    if user.is_empty() {
631        return None;
632    }
633    Some(percent_decode(user.as_bytes()).decode_utf8_lossy())
634}
635
636fn get_password_from_url(url: &Url) -> Option<Cow<'_, str>> {
637    let password = url.password()?;
638    Some(percent_decode(password.as_bytes()).decode_utf8_lossy())
639}
640
641fn get_database_from_url(url: &Url) -> Result<Option<&str>> {
642    match url.path_segments() {
643        None => Ok(None),
644        Some(mut segments) => {
645            let head = segments.next();
646
647            if segments.next().is_some() {
648                return Err(Error::Url(UrlError::Invalid));
649            }
650
651            match head {
652                Some(database) if !database.is_empty() => Ok(Some(database)),
653                _ => Ok(None),
654            }
655        }
656    }
657}
658
659fn parse_duration(source: &str) -> std::result::Result<Duration, ()> {
660    let digits_count = source.chars().take_while(|c| c.is_ascii_digit()).count();
661
662    let left: String = source.chars().take(digits_count).collect();
663    let right: String = source.chars().skip(digits_count).collect();
664
665    let num = match u64::from_str(&left) {
666        Ok(value) => value,
667        Err(_) => return Err(()),
668    };
669
670    match right.as_str() {
671        "s" => Ok(Duration::from_secs(num)),
672        "ms" => Ok(Duration::from_millis(num)),
673        _ => Err(()),
674    }
675}
676
677fn parse_opt_duration(source: &str) -> std::result::Result<Option<Duration>, ()> {
678    if source == "none" {
679        return Ok(None);
680    }
681
682    let duration = parse_duration(source)?;
683    Ok(Some(duration))
684}
685
686fn parse_compression(source: &str) -> std::result::Result<bool, ()> {
687    match source {
688        "none" => Ok(false),
689        "lz4" => Ok(true),
690        _ => Err(()),
691    }
692}
693
694fn parse_hosts(source: &str) -> std::result::Result<Vec<Url>, ()> {
695    let mut result = Vec::new();
696    for host in source.split(',') {
697        match Url::from_str(&format!("tcp://{host}")) {
698            Ok(url) => result.push(url),
699            Err(_) => return Err(()),
700        }
701    }
702    Ok(result)
703}
704
705#[cfg(test)]
706mod test {
707    use super::*;
708
709    #[test]
710    fn test_parse_hosts() {
711        let source = "host2:9000,host3:9000";
712        let expected = vec![
713            Url::from_str("tcp://host2:9000").unwrap(),
714            Url::from_str("tcp://host3:9000").unwrap(),
715        ];
716        let actual = parse_hosts(source).unwrap();
717        assert_eq!(actual, expected)
718    }
719
720    #[test]
721    fn test_parse_default() {
722        let url = "tcp://host1";
723        let options = from_url(url).unwrap();
724        assert_eq!(options.database, "default");
725        assert_eq!(options.username, "default");
726        assert_eq!(options.password, "");
727    }
728
729    #[test]
730    #[cfg(feature = "_tls")]
731    fn test_parse_secure_options() {
732        let url = "tcp://username:password@host1:9001/database?ping_timeout=42ms&keepalive=99s&compression=lz4&connection_timeout=10s&secure=true&skip_verify=true";
733        assert_eq!(
734            Options {
735                username: "username".into(),
736                password: "password".into(),
737                addr: Url::parse("tcp://username:password@host1:9001").unwrap(),
738                database: "database".into(),
739                keepalive: Some(Duration::from_secs(99)),
740                ping_timeout: Duration::from_millis(42),
741                connection_timeout: Duration::from_secs(10),
742                compression: true,
743                secure: true,
744                skip_verify: true,
745                ..Options::default()
746            },
747            from_url(url).unwrap(),
748        );
749    }
750
751    #[test]
752    fn test_parse_encoded_creds() {
753        let url = "tcp://user%20%3Cbar%3E:password%20%3Cbar%3E@host1:9001/database?ping_timeout=42ms&keepalive=99s&compression=lz4&connection_timeout=10s";
754        assert_eq!(
755            Options {
756                username: "user <bar>".into(),
757                password: "password <bar>".into(),
758                addr: Url::parse("tcp://user%20%3Cbar%3E:password%20%3Cbar%3E@host1:9001").unwrap(),
759                database: "database".into(),
760                keepalive: Some(Duration::from_secs(99)),
761                ping_timeout: Duration::from_millis(42),
762                connection_timeout: Duration::from_secs(10),
763                compression: true,
764                ..Options::default()
765            },
766            from_url(url).unwrap(),
767        );
768    }
769
770    #[test]
771    fn test_parse_options() {
772        let url = "tcp://username:password@host1:9001/database?ping_timeout=42ms&keepalive=99s&compression=lz4&connection_timeout=10s";
773        assert_eq!(
774            Options {
775                username: "username".into(),
776                password: "password".into(),
777                addr: Url::parse("tcp://username:password@host1:9001").unwrap(),
778                database: "database".into(),
779                keepalive: Some(Duration::from_secs(99)),
780                ping_timeout: Duration::from_millis(42),
781                connection_timeout: Duration::from_secs(10),
782                compression: true,
783                ..Options::default()
784            },
785            from_url(url).unwrap(),
786        );
787    }
788
789    #[test]
790    #[should_panic]
791    fn test_parse_invalid_url() {
792        let url = "ʘ_ʘ";
793        from_url(url).unwrap();
794    }
795
796    #[test]
797    fn test_parse_with_unknown_setting() {
798        let url = "tcp://localhost:9000/foo?bar=baz";
799        assert_eq!(
800            Options {
801                addr: Url::parse("tcp://localhost:9000").unwrap(),
802                database: "foo".into(),
803                settings: HashMap::from([(
804                    "bar".into(),
805                    SettingValue {
806                        value: SettingType::String("baz".into()),
807                        is_important: true,
808                    }
809                ),]),
810                ..Options::default()
811            },
812            from_url(url).unwrap(),
813        );
814        // TODO: try to run "SELECT 1" with it and got a failure
815    }
816
817    #[test]
818    fn test_with_setting() {
819        {
820            let opts = Options::from_str("tcp://localhost:9000")
821                .unwrap()
822                .with_setting("foo", "bar", true);
823            assert_eq!(
824                opts.settings,
825                HashMap::from([(
826                    "foo".into(),
827                    SettingValue {
828                        value: SettingType::String("bar".into()),
829                        is_important: true,
830                    }
831                )])
832            );
833        }
834
835        {
836            let opts = Options::from_str("tcp://localhost:9000")
837                .unwrap()
838                .with_setting("foo", "bar", false);
839            assert_eq!(
840                opts.settings,
841                HashMap::from([(
842                    "foo".into(),
843                    SettingValue {
844                        value: SettingType::String("bar".into()),
845                        is_important: false,
846                    }
847                )])
848            );
849        }
850
851        {
852            let opts = Options::from_str("tcp://localhost:9000")
853                .unwrap()
854                .with_setting("foo", 1, true);
855            assert_eq!(
856                opts.settings,
857                HashMap::from([(
858                    "foo".into(),
859                    SettingValue {
860                        value: SettingType::UInt64(1u64),
861                        is_important: true,
862                    }
863                )])
864            );
865        }
866
867        {
868            let opts = Options::from_str("tcp://localhost:9000")
869                .unwrap()
870                .with_setting("foo", true, true);
871            assert_eq!(
872                opts.settings,
873                HashMap::from([(
874                    "foo".into(),
875                    SettingValue {
876                        value: SettingType::Bool(true),
877                        is_important: true,
878                    }
879                )])
880            );
881        }
882
883        {
884            let opts = Options::from_str("tcp://localhost:9000")
885                .unwrap()
886                .with_setting("foo", 1., true);
887            assert_eq!(
888                opts.settings,
889                HashMap::from([(
890                    "foo".into(),
891                    SettingValue {
892                        value: SettingType::Float64(1.),
893                        is_important: true,
894                    }
895                )])
896            );
897        }
898    }
899
900    #[test]
901    #[should_panic]
902    fn test_parse_with_multi_databases() {
903        let url = "tcp://localhost:9000/foo/bar";
904        from_url(url).unwrap();
905    }
906
907    #[test]
908    fn test_parse_duration() {
909        assert_eq!(parse_duration("3s").unwrap(), Duration::from_secs(3));
910        assert_eq!(parse_duration("123ms").unwrap(), Duration::from_millis(123));
911
912        parse_duration("ms").unwrap_err();
913        parse_duration("1ss").unwrap_err();
914    }
915
916    #[test]
917    fn test_parse_opt_duration() {
918        assert_eq!(
919            parse_opt_duration("3s").unwrap(),
920            Some(Duration::from_secs(3))
921        );
922        assert_eq!(parse_opt_duration("none").unwrap(), None::<Duration>);
923    }
924
925    #[test]
926    fn test_parse_compression() {
927        assert!(!parse_compression("none").unwrap());
928        assert!(parse_compression("lz4").unwrap());
929        parse_compression("?").unwrap_err();
930    }
931}
clickhouse_rs_async/types/options.rs

clickhouse_rs_async/types/
options.rs
