risingwave_connector/source/mqtt/
mod.rs1pub mod enumerator;
16pub mod source;
17pub use enumerator::MqttSplitEnumerator;
18pub mod split;
19
20use std::collections::HashMap;
21use std::fmt::{Display, Formatter};
22
23use serde_derive::Deserialize;
24use serde_with::{DisplayFromStr, serde_as};
25use thiserror::Error;
26use with_options::WithOptions;
27
28use crate::connector_common::{MqttCommon, MqttQualityOfService};
29use crate::source::SourceProperties;
30use crate::source::mqtt::source::{MqttSplit, MqttSplitReader};
31
32pub const MQTT_CONNECTOR: &str = "mqtt";
33
34#[derive(Debug, Clone, Error)]
35pub struct MqttError(String);
36
37impl Display for MqttError {
38 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
39 write!(f, "{}", self.0)
40 }
41}
42
43#[serde_as]
44#[derive(Clone, Debug, Deserialize, WithOptions)]
45pub struct MqttProperties {
46 #[serde(flatten)]
47 pub common: MqttCommon,
48
49 pub topic: String,
51
52 #[serde_as(as = "Option<DisplayFromStr>")]
55 pub qos: Option<MqttQualityOfService>,
56
57 #[serde(flatten)]
58 pub unknown_fields: HashMap<String, String>,
59}
60
61impl SourceProperties for MqttProperties {
62 type Split = MqttSplit;
63 type SplitEnumerator = MqttSplitEnumerator;
64 type SplitReader = MqttSplitReader;
65
66 const SOURCE_NAME: &'static str = MQTT_CONNECTOR;
67}
68
69impl crate::source::UnknownFields for MqttProperties {
70 fn unknown_fields(&self) -> HashMap<String, String> {
71 self.unknown_fields.clone()
72 }
73}