faucet_common_pubsub/
config.rs

1//! Credential + connection configuration shared by the Pub/Sub source and
2//! sink. No I/O here — the client builder lives in `client.rs`.
3
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// How to authenticate with Google Cloud Pub/Sub.
8///
9/// Serializes as `{ type: <method>, config: { … } }` (adjacent tagging,
10/// snake_case discriminators) — the consistent auth wire shape shared by
11/// every faucet connector:
12/// `{ type: service_account_json_file, config: { path: "/run/secrets/sa.json" } }`.
13#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
14#[serde(tag = "type", content = "config", rename_all = "snake_case")]
15pub enum PubsubCredentials {
16    /// Application Default Credentials — honours
17    /// `GOOGLE_APPLICATION_CREDENTIALS`, `GOOGLE_APPLICATION_CREDENTIALS_JSON`,
18    /// gcloud user creds, and the GCE/GKE metadata server, in that order.
19    #[default]
20    ApplicationDefault,
21    /// Path to a service-account JSON key file on disk.
22    ServiceAccountJsonFile {
23        /// Filesystem path to the service-account key JSON.
24        path: String,
25    },
26    /// Service-account JSON key as an inline string. Pair with
27    /// `${env:GCP_SA_JSON}` / `${secret:…}` interpolation in CLI configs so
28    /// the key never sits in the config file verbatim.
29    ServiceAccountJsonInline {
30        /// The service-account key JSON document.
31        json: String,
32    },
33    /// No credentials. Use with the Pub/Sub emulator, which does not validate
34    /// bearer tokens — the SDK otherwise tries to fetch ADC tokens at startup
35    /// and fails in environments without GCP credentials.
36    Anonymous,
37}
38
39/// Connection settings shared by the Pub/Sub source and sink. Flattened into
40/// each connector config via `#[serde(flatten)]`, so `project_id` /
41/// `endpoint` / `emulator_host` / `credentials` appear at the config top
42/// level.
43#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
44pub struct PubsubConnection {
45    /// GCP project id that owns the topic / subscription. Required for real
46    /// Pub/Sub; the emulator infers it from `PUBSUB_PROJECT_ID` when unset.
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub project_id: Option<String>,
49    /// Override the Pub/Sub API endpoint (host:port or URL). Rarely needed —
50    /// prefer `emulator_host` for the emulator.
51    #[serde(default, skip_serializing_if = "Option::is_none")]
52    pub endpoint: Option<String>,
53    /// Point the client at a Pub/Sub emulator (`host:port`). When set, auth is
54    /// skipped and the endpoint is taken from this value — mirrors the
55    /// `PUBSUB_EMULATOR_HOST` environment variable the SDK honours.
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    pub emulator_host: Option<String>,
58    /// How to authenticate. Defaults to Application Default Credentials.
59    #[serde(default)]
60    pub credentials: PubsubCredentials,
61}
62
63impl PubsubConnection {
64    /// Effective emulator host: the explicit config value, else the
65    /// `PUBSUB_EMULATOR_HOST` environment variable. `None` = real Pub/Sub.
66    pub fn effective_emulator_host(&self) -> Option<String> {
67        self.emulator_host
68            .clone()
69            .or_else(|| std::env::var("PUBSUB_EMULATOR_HOST").ok())
70            .filter(|h| !h.trim().is_empty())
71    }
72}
73
74#[cfg(test)]
75mod tests {
76    use super::*;
77    use serde_json::json;
78
79    #[test]
80    fn credentials_default_is_adc() {
81        assert_eq!(
82            PubsubCredentials::default(),
83            PubsubCredentials::ApplicationDefault
84        );
85    }
86
87    #[test]
88    fn credentials_serde_application_default() {
89        let v = serde_json::to_value(PubsubCredentials::ApplicationDefault).unwrap();
90        assert_eq!(v, json!({"type": "application_default"}));
91        let back: PubsubCredentials = serde_json::from_value(v).unwrap();
92        assert_eq!(back, PubsubCredentials::ApplicationDefault);
93    }
94
95    #[test]
96    fn credentials_serde_service_account_file_and_inline() {
97        let file = PubsubCredentials::ServiceAccountJsonFile {
98            path: "/run/secrets/sa.json".into(),
99        };
100        let v = serde_json::to_value(&file).unwrap();
101        assert_eq!(
102            v,
103            json!({"type": "service_account_json_file", "config": {"path": "/run/secrets/sa.json"}})
104        );
105        assert_eq!(
106            serde_json::from_value::<PubsubCredentials>(v).unwrap(),
107            file
108        );
109
110        let inline = PubsubCredentials::ServiceAccountJsonInline {
111            json: "{\"client_email\":\"x@y\"}".into(),
112        };
113        let v = serde_json::to_value(&inline).unwrap();
114        assert_eq!(v["type"], "service_account_json_inline");
115        assert_eq!(
116            serde_json::from_value::<PubsubCredentials>(v).unwrap(),
117            inline
118        );
119    }
120
121    #[test]
122    fn credentials_serde_anonymous() {
123        let v = serde_json::to_value(PubsubCredentials::Anonymous).unwrap();
124        assert_eq!(v, json!({"type": "anonymous"}));
125        assert_eq!(
126            serde_json::from_value::<PubsubCredentials>(v).unwrap(),
127            PubsubCredentials::Anonymous
128        );
129    }
130
131    #[test]
132    fn connection_flatten_shape_parses() {
133        let yaml = r#"
134project_id: my-proj
135emulator_host: "localhost:8085"
136credentials: { type: anonymous }
137"#;
138        let c: PubsubConnection = serde_yaml::from_str(yaml).unwrap();
139        assert_eq!(c.project_id.as_deref(), Some("my-proj"));
140        assert_eq!(c.emulator_host.as_deref(), Some("localhost:8085"));
141        assert_eq!(c.credentials, PubsubCredentials::Anonymous);
142    }
143
144    #[test]
145    fn connection_defaults() {
146        let c = PubsubConnection::default();
147        assert!(c.project_id.is_none());
148        assert!(c.endpoint.is_none());
149        assert!(c.emulator_host.is_none());
150        assert_eq!(c.credentials, PubsubCredentials::ApplicationDefault);
151    }
152
153    #[test]
154    fn effective_emulator_host_prefers_explicit() {
155        let c = PubsubConnection {
156            emulator_host: Some("localhost:8085".into()),
157            ..Default::default()
158        };
159        assert_eq!(
160            c.effective_emulator_host().as_deref(),
161            Some("localhost:8085")
162        );
163
164        // Blank explicit value is ignored.
165        let c = PubsubConnection {
166            emulator_host: Some("   ".into()),
167            ..Default::default()
168        };
169        // Only asserts the blank-explicit path; env-var state is process-wide
170        // so we don't assert on its presence/absence here.
171        let got = c.effective_emulator_host();
172        assert!(got.as_deref() != Some("   "));
173    }
174}
faucet_common_pubsub/config.rs

faucet_common_pubsub/
config.rs
