risingwave_connector/source/pulsar/
mod.rs1pub mod enumerator;
16pub mod source;
17pub mod split;
18pub mod topic;
19
20use std::collections::HashMap;
21
22pub use enumerator::*;
23use serde::Deserialize;
24use serde_with::serde_as;
25pub use split::*;
26use with_options::WithOptions;
27
28use self::source::reader::PulsarSplitReader;
29use crate::connector_common::{AwsAuthProps, PulsarCommon, PulsarOauthCommon};
30use crate::source::SourceProperties;
31
32pub const PULSAR_CONNECTOR: &str = "pulsar";
33
34impl SourceProperties for PulsarProperties {
35 type Split = PulsarSplit;
36 type SplitEnumerator = PulsarSplitEnumerator;
37 type SplitReader = PulsarSplitReader;
38
39 const SOURCE_NAME: &'static str = PULSAR_CONNECTOR;
40}
41
42impl crate::source::UnknownFields for PulsarProperties {
43 fn unknown_fields(&self) -> HashMap<String, String> {
44 self.unknown_fields.clone()
45 }
46}
47
48#[derive(Clone, Debug, Deserialize, WithOptions)]
49#[serde_as]
50pub struct PulsarProperties {
51 #[serde(rename = "scan.startup.mode", alias = "pulsar.scan.startup.mode")]
52 pub scan_startup_mode: Option<String>,
53
54 #[serde(
55 rename = "scan.startup.timestamp.millis",
56 alias = "pulsar.time.offset",
57 alias = "scan.startup.timestamp_millis"
58 )]
59 pub time_offset: Option<String>,
60
61 #[serde(flatten)]
62 pub common: PulsarCommon,
63
64 #[serde(flatten)]
65 pub oauth: Option<PulsarOauthCommon>,
66
67 #[serde(flatten)]
68 pub aws_auth_props: AwsAuthProps,
69
70 #[serde(rename = "iceberg.enabled")]
71 #[serde_as(as = "DisplayFromStr")]
72 pub iceberg_loader_enabled: Option<bool>,
73
74 #[serde(rename = "iceberg.bucket", default)]
75 pub iceberg_bucket: Option<String>,
76
77 #[serde(rename = "subscription.name.prefix")]
85 pub subscription_name_prefix: Option<String>,
86
87 #[serde(flatten)]
88 pub unknown_fields: HashMap<String, String>,
89}