|
| 1 | +//! Pulsar Admin REST API client. |
| 2 | +//! |
| 3 | +//! Enabled by the `admin-api` feature flag. Requires a tokio runtime. |
| 4 | +
|
| 5 | +use std::sync::Arc; |
| 6 | + |
| 7 | +use futures::lock::Mutex; |
| 8 | + |
| 9 | +use crate::{ |
| 10 | + authentication::Authentication, |
| 11 | + connection_manager::TlsOptions, |
| 12 | + error::{AdminError, Error}, |
| 13 | +}; |
| 14 | + |
| 15 | +/// Parses a Pulsar topic URL into (scheme, tenant, namespace, topic_name). |
| 16 | +/// Accepts `persistent://` and `non-persistent://` prefixes, or a bare |
| 17 | +/// `tenant/namespace/topic` string which defaults to `persistent://`. |
| 18 | +fn parse_topic(topic: &str) -> Result<(&str, &str, &str, &str), Error> { |
| 19 | + let invalid = || { |
| 20 | + Error::Admin(AdminError::InvalidTopic(format!( |
| 21 | + "expected tenant/namespace/topic or a fully-qualified topic URL, got: {topic}" |
| 22 | + ))) |
| 23 | + }; |
| 24 | + |
| 25 | + let (scheme, rest) = if let Some(rest) = topic.strip_prefix("persistent://") { |
| 26 | + ("persistent", rest) |
| 27 | + } else if let Some(rest) = topic.strip_prefix("non-persistent://") { |
| 28 | + ("non-persistent", rest) |
| 29 | + } else { |
| 30 | + ("persistent", topic) |
| 31 | + }; |
| 32 | + |
| 33 | + let mut parts = rest.splitn(3, '/'); |
| 34 | + let tenant = parts.next().filter(|s| !s.is_empty()).ok_or_else(invalid)?; |
| 35 | + let namespace = parts.next().filter(|s| !s.is_empty()).ok_or_else(invalid)?; |
| 36 | + let name = parts.next().filter(|s| !s.is_empty()).ok_or_else(invalid)?; |
| 37 | + |
| 38 | + Ok((scheme, tenant, namespace, name)) |
| 39 | +} |
| 40 | + |
| 41 | +/// Client for the Pulsar Admin REST API. |
| 42 | +/// |
| 43 | +/// Obtain an instance via [`Pulsar::admin()`][crate::Pulsar::admin]. |
| 44 | +/// |
| 45 | +/// # Example |
| 46 | +/// |
| 47 | +/// ```rust,no_run |
| 48 | +/// # async fn run(pulsar: pulsar::Pulsar<pulsar::TokioExecutor>) -> Result<(), pulsar::Error> { |
| 49 | +/// let admin = pulsar.admin("http://localhost:8080")?; |
| 50 | +/// admin |
| 51 | +/// .set_max_unacked_messages_per_consumer( |
| 52 | +/// "persistent://public/default/my-topic", |
| 53 | +/// 500, |
| 54 | +/// ) |
| 55 | +/// .await?; |
| 56 | +/// # Ok(()) |
| 57 | +/// # } |
| 58 | +/// ``` |
| 59 | +pub struct AdminClient { |
| 60 | + client: reqwest::Client, |
| 61 | + admin_url: String, |
| 62 | + auth: Option<Arc<Mutex<Box<dyn Authentication>>>>, |
| 63 | +} |
| 64 | + |
| 65 | +impl AdminClient { |
| 66 | + /// Creates a new `AdminClient`. |
| 67 | + /// |
| 68 | + /// Reuses the TLS and authentication configuration already present on the |
| 69 | + /// [`Pulsar`][crate::Pulsar] client. Called internally by |
| 70 | + /// [`Pulsar::admin()`][crate::Pulsar::admin]. |
| 71 | + pub(crate) fn new( |
| 72 | + admin_url: String, |
| 73 | + tls_options: &TlsOptions, |
| 74 | + auth: Option<Arc<Mutex<Box<dyn Authentication>>>>, |
| 75 | + ) -> Result<Self, Error> { |
| 76 | + let mut builder = reqwest::ClientBuilder::new() |
| 77 | + .danger_accept_invalid_certs(tls_options.allow_insecure_connection); |
| 78 | + |
| 79 | + builder = builder.danger_accept_invalid_hostnames( |
| 80 | + tls_options.allow_insecure_connection || !tls_options.tls_hostname_verification_enabled, |
| 81 | + ); |
| 82 | + |
| 83 | + if let Some(pem_bytes) = &tls_options.certificate_chain { |
| 84 | + let certs = pem::parse_many(pem_bytes) |
| 85 | + .map_err(|e| Error::Custom(format!("failed to parse certificate chain: {e}")))?; |
| 86 | + for cert in certs.iter().rev() { |
| 87 | + let reqwest_cert = reqwest::Certificate::from_der(cert.contents()) |
| 88 | + .map_err(|e| Error::Admin(AdminError::Request(e)))?; |
| 89 | + builder = builder.add_root_certificate(reqwest_cert); |
| 90 | + } |
| 91 | + } |
| 92 | + |
| 93 | + Ok(AdminClient { |
| 94 | + client: builder |
| 95 | + .build() |
| 96 | + .map_err(|e| Error::Admin(AdminError::Request(e)))?, |
| 97 | + admin_url: admin_url.trim_end_matches('/').to_string(), |
| 98 | + auth, |
| 99 | + }) |
| 100 | + } |
| 101 | + |
| 102 | + async fn apply_auth( |
| 103 | + &self, |
| 104 | + req: reqwest::RequestBuilder, |
| 105 | + ) -> Result<reqwest::RequestBuilder, Error> { |
| 106 | + let Some(auth) = &self.auth else { |
| 107 | + return Ok(req); |
| 108 | + }; |
| 109 | + let mut auth = auth.lock().await; |
| 110 | + let method = auth.auth_method_name(); |
| 111 | + let data = auth.auth_data().await.map_err(Error::Authentication)?; |
| 112 | + let data_str = String::from_utf8(data) |
| 113 | + .map_err(|e| Error::Custom(format!("auth data is not valid UTF-8: {e}")))?; |
| 114 | + Ok(match method.as_str() { |
| 115 | + "token" => req.bearer_auth(data_str), |
| 116 | + "basic" => match data_str.split_once(':') { |
| 117 | + Some((user, pass)) => req.basic_auth(user, Some(pass)), |
| 118 | + None => req.basic_auth(&data_str, None::<&str>), |
| 119 | + }, |
| 120 | + _ => req, |
| 121 | + }) |
| 122 | + } |
| 123 | + |
| 124 | + fn topic_policy_url(&self, topic: &str, policy: &str) -> Result<String, Error> { |
| 125 | + let (scheme, tenant, namespace, name) = parse_topic(topic)?; |
| 126 | + Ok(format!( |
| 127 | + "{}/admin/v2/{}/{}/{}/{}/{policy}", |
| 128 | + self.admin_url, scheme, tenant, namespace, name |
| 129 | + )) |
| 130 | + } |
| 131 | + |
| 132 | + async fn check_response(&self, resp: reqwest::Response) -> Result<(), Error> { |
| 133 | + if resp.status().is_success() { |
| 134 | + return Ok(()); |
| 135 | + } |
| 136 | + let status = resp.status().as_u16(); |
| 137 | + let body = resp.text().await.unwrap_or_default(); |
| 138 | + Err(Error::Admin(AdminError::Http { status, body })) |
| 139 | + } |
| 140 | + |
| 141 | + /// Sets the maximum number of unacknowledged messages allowed per consumer |
| 142 | + /// on a topic. |
| 143 | + /// |
| 144 | + /// This is a persistent broker-side topic policy. The topic must already |
| 145 | + /// exist when this is called (subscribe a consumer first, then call this). |
| 146 | + /// Requires `topicLevelPoliciesEnabled=true` in the broker configuration. |
| 147 | + pub async fn set_max_unacked_messages_per_consumer( |
| 148 | + &self, |
| 149 | + topic: &str, |
| 150 | + max_unacked: u32, |
| 151 | + ) -> Result<(), Error> { |
| 152 | + let url = self.topic_policy_url(topic, "maxUnackedMessagesOnConsumer")?; |
| 153 | + let req = self |
| 154 | + .client |
| 155 | + .post(&url) |
| 156 | + .header("Content-Type", "application/json") |
| 157 | + .body(max_unacked.to_string()); |
| 158 | + let req = self.apply_auth(req).await?; |
| 159 | + let resp = req |
| 160 | + .send() |
| 161 | + .await |
| 162 | + .map_err(|e| Error::Admin(AdminError::Request(e)))?; |
| 163 | + self.check_response(resp).await |
| 164 | + } |
| 165 | + |
| 166 | + /// Removes the per-topic max unacked messages override, reverting to the |
| 167 | + /// broker or namespace default. |
| 168 | + /// |
| 169 | + /// To disable the limit without removing the topic-level override, call |
| 170 | + /// [`set_max_unacked_messages_per_consumer`][Self::set_max_unacked_messages_per_consumer] |
| 171 | + /// with a value of `0` (unlimited). |
| 172 | + pub async fn remove_max_unacked_messages_per_consumer(&self, topic: &str) -> Result<(), Error> { |
| 173 | + let url = self.topic_policy_url(topic, "maxUnackedMessagesOnConsumer")?; |
| 174 | + let req = self.client.delete(&url); |
| 175 | + let req = self.apply_auth(req).await?; |
| 176 | + let resp = req |
| 177 | + .send() |
| 178 | + .await |
| 179 | + .map_err(|e| Error::Admin(AdminError::Request(e)))?; |
| 180 | + self.check_response(resp).await |
| 181 | + } |
| 182 | +} |
| 183 | + |
| 184 | +#[cfg(test)] |
| 185 | +mod tests { |
| 186 | + use super::*; |
| 187 | + |
| 188 | + #[test] |
| 189 | + fn test_parse_topic_persistent() { |
| 190 | + let (scheme, tenant, ns, name) = |
| 191 | + parse_topic("persistent://my-tenant/my-namespace/my-topic").unwrap(); |
| 192 | + assert_eq!(scheme, "persistent"); |
| 193 | + assert_eq!(tenant, "my-tenant"); |
| 194 | + assert_eq!(ns, "my-namespace"); |
| 195 | + assert_eq!(name, "my-topic"); |
| 196 | + } |
| 197 | + |
| 198 | + #[test] |
| 199 | + fn test_parse_topic_non_persistent() { |
| 200 | + let (scheme, tenant, ns, name) = parse_topic("non-persistent://tenant/ns/topic").unwrap(); |
| 201 | + assert_eq!(scheme, "non-persistent"); |
| 202 | + assert_eq!(tenant, "tenant"); |
| 203 | + assert_eq!(ns, "ns"); |
| 204 | + assert_eq!(name, "topic"); |
| 205 | + } |
| 206 | + |
| 207 | + #[test] |
| 208 | + fn test_parse_topic_bare() { |
| 209 | + // No prefix defaults to persistent:// |
| 210 | + let (scheme, tenant, ns, name) = parse_topic("tenant/ns/topic").unwrap(); |
| 211 | + assert_eq!(scheme, "persistent"); |
| 212 | + assert_eq!(tenant, "tenant"); |
| 213 | + assert_eq!(ns, "ns"); |
| 214 | + assert_eq!(name, "topic"); |
| 215 | + } |
| 216 | + |
| 217 | + #[test] |
| 218 | + fn test_parse_topic_missing_parts() { |
| 219 | + assert!(parse_topic("").is_err()); |
| 220 | + assert!(parse_topic("tenant").is_err()); |
| 221 | + assert!(parse_topic("tenant/ns").is_err()); |
| 222 | + // trailing slash = empty topic name |
| 223 | + assert!(parse_topic("tenant/ns/").is_err()); |
| 224 | + assert!(parse_topic("persistent://").is_err()); |
| 225 | + assert!(parse_topic("persistent://tenant").is_err()); |
| 226 | + assert!(parse_topic("persistent://tenant/ns").is_err()); |
| 227 | + assert!(parse_topic("persistent://tenant/ns/").is_err()); |
| 228 | + } |
| 229 | + |
| 230 | + #[test] |
| 231 | + fn test_topic_policy_url() { |
| 232 | + let client = AdminClient { |
| 233 | + client: reqwest::Client::new(), |
| 234 | + admin_url: "http://localhost:8080".to_string(), |
| 235 | + auth: None, |
| 236 | + }; |
| 237 | + assert_eq!( |
| 238 | + client |
| 239 | + .topic_policy_url( |
| 240 | + "persistent://public/default/my-topic", |
| 241 | + "maxUnackedMessagesOnConsumer" |
| 242 | + ) |
| 243 | + .unwrap(), |
| 244 | + "http://localhost:8080/admin/v2/persistent/public/default/my-topic/maxUnackedMessagesOnConsumer" |
| 245 | + ); |
| 246 | + } |
| 247 | + |
| 248 | + #[test] |
| 249 | + fn test_admin_url_trailing_slash_stripped() { |
| 250 | + // Trailing slash on admin_url should be normalized away |
| 251 | + let tls = TlsOptions::default(); |
| 252 | + let client = AdminClient::new("http://localhost:8080/".to_string(), &tls, None).unwrap(); |
| 253 | + assert_eq!(client.admin_url, "http://localhost:8080"); |
| 254 | + } |
| 255 | +} |
0 commit comments