Skip to content

Commit

Permalink
Add service metadata
Browse files Browse the repository at this point in the history 
Signed-off-by: Tomasz Pietrek <tomasz@nats.io>
  • Loading branch information
@Jarema
Jarema committed May 11, 2023
1 parent 3765431 commit 1c0f663
Show file tree
Hide file tree
Showing 2 changed files with 68 additions and 4 deletions.
22 changes: 20 additions & 2 deletions async-nats/src/service/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,8 @@ pub struct Info {
pub version: String,
/// All service endpoints.
pub subjects: Vec<String>,
/// Additional metadata
pub metadata: HashMap<String, String>,
}

/// Schema of requests and responses.
Expand All @@ -145,13 +147,16 @@ pub struct Config {
pub schema: Option<Schema>,
/// Custom handler for providing the `EndpointStats.data` value.
pub stats_handler: Option<StatsHandler>,
/// Additional service metadata
pub metadata: Option<HashMap<String, String>>,
}

pub struct ServiceBuilder {
client: Client,
description: Option<String>,
schema: Option<Schema>,
stats_handler: Option<StatsHandler>,
metadata: Option<HashMap<String, String>>,
}

impl ServiceBuilder {
Expand All @@ -161,6 +166,7 @@ impl ServiceBuilder {
description: None,
schema: None,
stats_handler: None,
metadata: None,
}
}

Expand All @@ -185,6 +191,12 @@ impl ServiceBuilder {
self
}

/// Adds additional metadata.
pub fn metadata(mut self, metadata: HashMap<String, String>) -> Self {
self.metadata = Some(metadata);
self
}

/// Stats the service with configured options.
pub async fn start<S: ToString>(self, name: S, version: S) -> Result<Service, Error> {
Service::add(
Expand All @@ -195,6 +207,7 @@ impl ServiceBuilder {
description: self.description,
schema: self.schema,
stats_handler: self.stats_handler,
metadata: self.metadata,
},
)
.await
Expand Down Expand Up @@ -240,6 +253,7 @@ pub trait ServiceExt {
/// schema: None,
/// description: None,
/// stats_handler: None,
/// metadata: None,
/// })
/// .await?;
///
Expand Down Expand Up @@ -312,6 +326,7 @@ impl ServiceExt for crate::Client {
/// schema: None,
/// description: None,
/// stats_handler: None,
/// metadata: None,
/// })
/// .await?;
///
Expand Down Expand Up @@ -416,6 +431,7 @@ impl Service {
description: config.description.clone(),
version: config.version.clone(),
subjects: Vec::default(),
metadata: config.metadata.clone().unwrap_or_default(),
};

let (shutdown_tx, _) = tokio::sync::broadcast::channel(1);
Expand Down Expand Up @@ -450,6 +466,7 @@ impl Service {
"name": config.name.clone(),
"id": id.clone(),
"version": config.version.clone(),
"metadata": config.metadata.clone(),
}))
.map(Bytes::from)?;
async move {
Expand Down Expand Up @@ -698,15 +715,16 @@ impl Request {
/// ```no_run
/// # #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use futures::StreamExt;
/// use async_nats::service::ServiceExt;
/// use futures::StreamExt;
/// # let client = async_nats::connect("demo.nats.io").await?;
/// # let mut service = client.add_service(async_nats::service::Config {
/// # name: "generator".to_string(),
/// # version: "1.0.0".to_string(),
/// # schema: None,
/// # description: None,
/// stats_handler: None,
/// # stats_handler: None,
/// # metadata: None,
/// # }).await?;
///
/// let mut endpoint = service.endpoint("endpoint").await?;
Expand Down
50 changes: 48 additions & 2 deletions async-nats/tests/service_tests.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,9 @@

#[cfg(feature = "service")]
mod service {
use std::str::from_utf8;
use std::{collections::HashMap, str::from_utf8};

use async_nats::service::{Info, ServiceExt, StatsResponse};
use async_nats::service::{self, Info, ServiceExt, StatsResponse};
use futures::StreamExt;
use tracing::debug;

Expand All @@ -31,6 +31,7 @@ mod service {
version: "1.0.0.1".to_string(),
schema: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap_err()
Expand All @@ -47,6 +48,7 @@ mod service {
version: "beta-1.0.0".to_string(),
schema: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap_err()
Expand All @@ -63,6 +65,7 @@ mod service {
version: "1.0.0".to_string(),
schema: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap_err()
Expand All @@ -79,6 +82,7 @@ mod service {
version: "1.0.0".to_string(),
schema: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap_err()
Expand All @@ -88,6 +92,44 @@ mod service {
assert_eq!(std::io::ErrorKind::InvalidInput, err_kind);
}

#[tokio::test]
async fn metadata() {
let server = nats_server::run_basic_server();
let client = async_nats::connect(server.client_url()).await.unwrap();
let metadata = HashMap::from([
("key".to_string(), "value".to_string()),
("other".to_string(), "value".to_string()),
]);
client
.add_service(async_nats::service::Config {
name: "serviceA".to_string(),
description: None,
version: "1.0.0".to_string(),
schema: None,
stats_handler: None,
metadata: Some(metadata.clone()),
})
.await
.unwrap();

let reply = client.new_inbox();
let mut responses = client.subscribe(reply.clone()).await.unwrap();
client
.publish_with_reply("$SRV.INFO".to_string(), reply, "".into())
.await
.unwrap();
let response = responses
.next()
.await
.map(|message| {
serde_json::from_slice::<service::Info>(&message.payload)
.unwrap()
.metadata
})
.unwrap();
assert_eq!(metadata, response);
}

#[tokio::test]
async fn ping() {
let server = nats_server::run_basic_server();
Expand All @@ -99,6 +141,7 @@ mod service {
version: "1.0.0".to_string(),
schema: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap();
Expand All @@ -110,6 +153,7 @@ mod service {
version: "2.0.0".to_string(),
schema: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap();
Expand All @@ -136,6 +180,7 @@ mod service {
schema: None,
description: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap();
Expand Down Expand Up @@ -175,6 +220,7 @@ mod service {
schema: None,
description: None,
stats_handler: None,
metadata: None,
})
.await
.unwrap();
Expand Down

0 comments on commit 1c0f663

Please sign in to comment.