sea_streamer_socket/
connect_options.rs

1#[cfg(feature = "backend-file")]
2use sea_streamer_file::FileConnectOptions;
3#[cfg(feature = "backend-kafka")]
4use sea_streamer_kafka::KafkaConnectOptions;
5#[cfg(feature = "backend-redis")]
6use sea_streamer_redis::RedisConnectOptions;
7#[cfg(feature = "backend-stdio")]
8use sea_streamer_stdio::StdioConnectOptions;
9
10use crate::{map_err, BackendErr, SeaResult};
11use sea_streamer_types::ConnectOptions;
12use std::time::Duration;
13
14#[derive(Debug, Default, Clone)]
15/// `sea-streamer-socket` concrete type of ConnectOptions.
16pub struct SeaConnectOptions {
17    #[cfg(feature = "backend-kafka")]
18    kafka: KafkaConnectOptions,
19    #[cfg(feature = "backend-redis")]
20    redis: RedisConnectOptions,
21    #[cfg(feature = "backend-stdio")]
22    stdio: StdioConnectOptions,
23    #[cfg(feature = "backend-file")]
24    file: FileConnectOptions,
25}
26
27impl SeaConnectOptions {
28    #[cfg(feature = "backend-kafka")]
29    pub fn into_kafka_connect_options(self) -> KafkaConnectOptions {
30        self.kafka
31    }
32
33    #[cfg(feature = "backend-redis")]
34    pub fn into_redis_connect_options(self) -> RedisConnectOptions {
35        self.redis
36    }
37
38    #[cfg(feature = "backend-stdio")]
39    pub fn into_stdio_connect_options(self) -> StdioConnectOptions {
40        self.stdio
41    }
42
43    #[cfg(feature = "backend-file")]
44    pub fn into_file_connect_options(self) -> FileConnectOptions {
45        self.file
46    }
47
48    #[cfg(feature = "backend-kafka")]
49    /// Set options that only applies to Kafka
50    pub fn set_kafka_connect_options<F: FnOnce(&mut KafkaConnectOptions)>(&mut self, func: F) {
51        func(&mut self.kafka)
52    }
53
54    #[cfg(feature = "backend-redis")]
55    /// Set options that only applies to Redis
56    pub fn set_redis_connect_options<F: FnOnce(&mut RedisConnectOptions)>(&mut self, func: F) {
57        func(&mut self.redis)
58    }
59
60    #[cfg(feature = "backend-stdio")]
61    /// Set options that only applies to Stdio
62    pub fn set_stdio_connect_options<F: FnOnce(&mut StdioConnectOptions)>(&mut self, func: F) {
63        func(&mut self.stdio)
64    }
65
66    #[cfg(feature = "backend-file")]
67    /// Set options that only applies to File
68    pub fn set_file_connect_options<F: FnOnce(&mut FileConnectOptions)>(&mut self, func: F) {
69        func(&mut self.file)
70    }
71}
72
73impl ConnectOptions for SeaConnectOptions {
74    type Error = BackendErr;
75
76    fn timeout(&self) -> SeaResult<Duration> {
77        #![allow(unreachable_code)]
78
79        #[cfg(feature = "backend-kafka")]
80        return self.kafka.timeout().map_err(map_err);
81        #[cfg(feature = "backend-redis")]
82        return self.redis.timeout().map_err(map_err);
83        #[cfg(feature = "backend-stdio")]
84        return self.stdio.timeout().map_err(map_err);
85        #[cfg(feature = "backend-file")]
86        return self.file.timeout().map_err(map_err);
87    }
88
89    fn set_timeout(&mut self, d: Duration) -> SeaResult<&mut Self> {
90        #[cfg(feature = "backend-kafka")]
91        self.kafka.set_timeout(d).map_err(map_err)?;
92        #[cfg(feature = "backend-redis")]
93        self.redis.set_timeout(d).map_err(map_err)?;
94        #[cfg(feature = "backend-stdio")]
95        self.stdio.set_timeout(d).map_err(map_err)?;
96        #[cfg(feature = "backend-file")]
97        self.file.set_timeout(d).map_err(map_err)?;
98
99        Ok(self)
100    }
101}
sea_streamer_socket/connect_options.rs

sea_streamer_socket/
connect_options.rs
