diff options
| -rw-r--r-- | Cargo.toml | 3 | ||||
| -rw-r--r-- | examples/test.rs | 15 | ||||
| -rw-r--r-- | src/catalog.rs | 20 | ||||
| -rw-r--r-- | src/kv.rs | 38 | ||||
| -rw-r--r-- | src/with_index.rs | 31 |
5 files changed, 82 insertions, 25 deletions
@@ -2,7 +2,7 @@ name = "df-consul" description = "Deuxfleurs' async Rust bindings for (a subset of) the Consul HTTP API" authors = [ "Alex Auvolat <alex@adnab.me>" ] -version = "0.3.3" +version = "0.3.4" edition = "2021" license = "MIT" repository = "https://git.deuxfleurs.fr/Deuxfleurs/df-consul" @@ -14,6 +14,7 @@ anyhow = "1.0.66" serde = { version = "1.0.149", features = ["derive"] } log = "0.4" bytes = "1" +base64 = "0.13" reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls-manual-roots" ] } tokio = { version = "1.23", default-features = false, features = [ "macros" ] } futures = "0.3.25" diff --git a/examples/test.rs b/examples/test.rs index 7e38c66..111c8c8 100644 --- a/examples/test.rs +++ b/examples/test.rs @@ -45,4 +45,19 @@ async fn main() { .unwrap() ); } + + println!("== LIST PREFIX =="); + let prefix = consul + .kv_get_prefix("diplonat/autodiscovery", None) + .await + .unwrap(); + println!("{:?}", prefix); + for i in 0..3 { + println!("-- wait for update... --"); + let prefix = consul + .kv_get_prefix("diplonat/autodiscovery", Some(prefix.index())) + .await + .unwrap(); + println!("{:?}", prefix); + } } diff --git a/src/catalog.rs b/src/catalog.rs index 042f745..fe82c5a 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -5,7 +5,6 @@ use std::cmp; use std::collections::HashMap; -use std::fmt::Write; use std::sync::Arc; use std::time::Duration; @@ -178,25 +177,6 @@ impl Consul { rx } - - async fn get_with_index<T: for<'de> Deserialize<'de>>( - &self, - mut url: String, - last_index: Option<usize>, - ) -> Result<WithIndex<T>> { - if let Some(i) = last_index { - if url.contains('?') { - write!(&mut url, "&index={}", i).unwrap(); - } else { - write!(&mut url, "?index={}", i).unwrap(); - } - } - debug!("GET {} as {}", url, std::any::type_name::<T>()); - - let http = self.client.get(&url).send().await?; - - Ok(WithIndex::<T>::index_from(&http)?.value(http.json().await?)) - } } async fn do_watch_all_service_health( @@ -1,10 +1,19 @@ +use std::collections::HashMap; + use anyhow::{anyhow, Result}; use bytes::Bytes; use log::*; use reqwest::StatusCode; use serde::{Deserialize, Serialize}; -use crate::Consul; +use crate::{Consul, WithIndex}; + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +#[serde(rename_all = "PascalCase")] +pub struct KvGetPrefixEntry { + pub key: String, + pub value: String, +} impl Consul { pub async fn kv_get(&self, key: &str) -> Result<Option<Bytes>> { @@ -37,6 +46,33 @@ impl Consul { } } + pub async fn kv_get_prefix( + &self, + key_prefix: &str, + last_index: Option<usize>, + ) -> Result<WithIndex<HashMap<String, Bytes>>> { + debug!("kv_get_prefix {} index={:?}", key_prefix, last_index); + let results: WithIndex<Vec<KvGetPrefixEntry>> = self + .get_with_index( + format!( + "{}/v1/kv/{}{}?recurse", + self.url, self.kv_prefix, key_prefix + ), + last_index, + ) + .await?; + + let mut res = HashMap::new(); + for ent in results.value { + res.insert(ent.key, Bytes::from(base64::decode(&ent.value)?)); + } + + Ok(WithIndex { + value: res, + index: results.index, + }) + } + pub async fn kv_put(&self, key: &str, bytes: Bytes) -> Result<()> { debug!("kv_put {}", key); diff --git a/src/with_index.rs b/src/with_index.rs index adce169..ba94779 100644 --- a/src/with_index.rs +++ b/src/with_index.rs @@ -1,14 +1,39 @@ -use std::fmt::{Debug, Display}; +use std::fmt::{Debug, Display, Write}; use anyhow::{bail, Result}; +use log::*; use reqwest::Response; +use serde::Deserialize; + +use crate::Consul; + +impl Consul { + pub(crate) async fn get_with_index<T: for<'de> Deserialize<'de>>( + &self, + mut url: String, + last_index: Option<usize>, + ) -> Result<WithIndex<T>> { + if let Some(i) = last_index { + if url.contains('?') { + write!(&mut url, "&index={}", i).unwrap(); + } else { + write!(&mut url, "?index={}", i).unwrap(); + } + } + debug!("GET {} as {}", url, std::any::type_name::<T>()); + + let http = self.client.get(&url).send().await?; + + Ok(WithIndex::<T>::index_from(&http)?.value(http.json().await?)) + } +} /// Wraps the returned value of an [API call with blocking /// possibility](https://developer.hashicorp.com/consul/api-docs/features/blocking) with the /// returned Consul index pub struct WithIndex<T> { - value: T, - index: usize, + pub(crate) value: T, + pub(crate) index: usize, } impl<T> WithIndex<T> { |