Skip to content

Commit

Permalink
Remove schema and api_url from Service API
Browse files Browse the repository at this point in the history 
This addresses nats-io/nats-architecture-and-design#220

Signed-off-by: Tomasz Pietrek <[email protected]>
  • Loading branch information
@Jarema
Jarema committed May 23, 2023
1 parent fc81ba6 commit d965068
Show file tree
Hide file tree
Showing 3 changed files with 0 additions and 119 deletions.
12 changes: 0 additions & 12 deletions async-nats/src/service/endpoint.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -131,8 +131,6 @@ pub(crate) struct Inner {
pub(crate) last_error: Option<error::Error>,
/// Custom data added by [Config::stats_handler]
pub(crate) data: String,
/// EndpointSchema
pub(crate) schema: Option<Schema>,
}

impl From<Inner> for Stats {
Expand All @@ -152,16 +150,6 @@ impl From<Inner> for Stats {
}
}

/// Schema of requests and responses.
/// Currently, it does not do anything except providing information.
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
pub struct Schema {
/// A string/url describing the format of the request payload can be JSON schema etc.
pub request: String,
/// A string/url describing the format of the request payload can be JSON schema etc.
pub response: String,
}

#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct Stats {
// Response type.
Expand Down
79 changes: 0 additions & 79 deletions async-nats/src/service/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@ use time::serde::rfc3339;
use time::OffsetDateTime;
use tokio::{sync::broadcast::Sender, task::JoinHandle};
use tracing::debug;
use url::Url;

use crate::{Client, Error, HeaderMap, Message, PublishError, Subscriber};

Expand Down Expand Up @@ -114,16 +113,13 @@ pub struct Config {
pub stats_handler: Option<StatsHandler>,
/// Additional service metadata
pub metadata: Option<HashMap<String, String>>,
/// A valid URL pointing to API specification (may contain schemas, paths etc.)
pub api_url: Option<url::Url>,
}

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

impl ServiceBuilder {
Expand All @@ -133,7 +129,6 @@ impl ServiceBuilder {
description: None,
stats_handler: None,
metadata: None,
api_url: None,
}
}

Expand All @@ -158,12 +153,6 @@ impl ServiceBuilder {
self
}

/// API URL.
pub fn api_url(mut self, api_url: Url) -> Self {
self.api_url = Some(api_url);
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 @@ -174,7 +163,6 @@ impl ServiceBuilder {
description: self.description,
stats_handler: self.stats_handler,
metadata: self.metadata,
api_url: self.api_url,
},
)
.await
Expand Down Expand Up @@ -220,7 +208,6 @@ pub trait ServiceExt {
/// description: None,
/// stats_handler: None,
/// metadata: None,
/// api_url: None,
/// })
/// .await?;
///
Expand Down Expand Up @@ -293,7 +280,6 @@ impl ServiceExt for crate::Client {
/// description: None,
/// stats_handler: None,
/// metadata: None,
/// api_url: None,
/// })
/// .await?;
///
Expand Down Expand Up @@ -355,13 +341,6 @@ impl Service {
verb_subscription(client.clone(), Verb::Ping, config.name.clone(), id.clone()).await?;
let mut infos =
verb_subscription(client.clone(), Verb::Info, config.name.clone(), id.clone()).await?;
let mut schemas = verb_subscription(
client.clone(),
Verb::Schema,
config.name.clone(),
id.clone(),
)
.await?;
let mut stats =
verb_subscription(client.clone(), Verb::Stats, config.name.clone(), id.clone()).await?;

Expand All @@ -372,7 +351,6 @@ impl Service {
let subjects = subjects.clone();
let endpoints_state = endpoints_state.clone();
let client = client.clone();
let api_url = config.api_url;
async move {
loop {
tokio::select! {
Expand All @@ -395,30 +373,6 @@ impl Service {
let info_json = serde_json::to_vec(&info).map(Bytes::from)?;
client.publish(info_request.reply.unwrap(), info_json.clone()).await?;
},
Some(schema_request) = schemas.next() => {
let endpoints_schema: Vec<EndpointSchema> = endpoints_state
.lock()
.unwrap()
.endpoints
.iter_mut()
.map(|(k, v)| EndpointSchema {
name: k.to_owned(),
subject: v.subject.to_owned(),
metadata: v.metadata.clone(),
schema: v.schema.clone(),
})
.collect();
let schema_json = serde_json::to_vec(&Schema {
kind: "io.nats.micro.v1.schema".to_string(),
name: info.name.clone(),
id: info.id.clone(),
version: info.version.clone(),
api_url: api_url.clone(),
endpoints: endpoints_schema,
})
.map(Bytes::from)?;
client.publish(schema_request.reply.unwrap(), schema_json.clone()).await?;
},
Some(stats_request) = stats.next() => {
if let Some(stats_callback) = stats_callback.as_mut() {
let mut endpoint_stats_locked = endpoints_state.lock().unwrap();
Expand Down Expand Up @@ -706,7 +660,6 @@ impl Request {
/// # description: None,
/// # stats_handler: None,
/// # metadata: None,
/// api_url: None,
/// # }).await?;
///
/// let mut endpoint = service.endpoint("endpoint").await?;
Expand Down Expand Up @@ -756,7 +709,6 @@ pub struct EndpointBuilder {
shutdown_tx: Sender<()>,
name: Option<String>,
metadata: Option<HashMap<String, String>>,
schema: Option<endpoint::Schema>,
subjects: Arc<Mutex<Vec<String>>>,
}

Expand All @@ -774,7 +726,6 @@ impl EndpointBuilder {
shutdown_tx,
name: None,
metadata: None,
schema: None,
}
}

Expand All @@ -790,15 +741,6 @@ impl EndpointBuilder {
self
}

/// [Schema] for the [Endpoint].
pub fn schema<S: ToString>(mut self, request: S, response: S) -> EndpointBuilder {
self.schema = Some(endpoint::Schema {
request: request.to_string(),
response: response.to_string(),
});
self
}

/// Finalizes the builder and adds the [Endpoint].
pub async fn add<S: ToString>(self, subject: S) -> Result<Endpoint, Error> {
let subject = subject.to_string();
Expand All @@ -818,7 +760,6 @@ impl EndpointBuilder {
.or_insert(endpoint::Inner {
name,
metadata: self.metadata.unwrap_or_default(),
schema: self.schema,
..Default::default()
});
self.subjects.lock().unwrap().push(subject.clone());
Expand All @@ -833,26 +774,6 @@ impl EndpointBuilder {
}
}

#[derive(Deserialize, Serialize, Debug)]
pub struct Schema {
#[serde(rename = "type")]
pub kind: String,
pub name: String,
pub id: String,
pub version: String,
#[serde(default)]
pub api_url: Option<Url>,
pub endpoints: Vec<EndpointSchema>,
}

#[derive(Deserialize, Serialize, Debug)]
pub struct EndpointSchema {
pub name: String,
pub subject: String,
pub metadata: HashMap<String, String>,
pub schema: Option<endpoint::Schema>,
}

pub struct StatsHandler(pub Box<dyn FnMut(String, endpoint::Stats) -> String + Send>);

impl std::fmt::Debug for StatsHandler {
Expand Down
28 changes: 0 additions & 28 deletions async-nats/tests/service_tests.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ mod service {
use async_nats::service::{self, Info, ServiceExt, Stats};
use futures::StreamExt;
use tracing::debug;
use url::Url;

#[tokio::test]
async fn service_config_validations() {
Expand Down Expand Up @@ -54,7 +53,6 @@ mod service {
version: "1.0.0".to_string(),
stats_handler: None,
metadata: None,
api_url: None,
})
.await
.unwrap_err()
Expand All @@ -71,7 +69,6 @@ mod service {
version: "1.0.0".to_string(),
stats_handler: None,
metadata: None,
api_url: None,
})
.await
.unwrap_err()
Expand All @@ -94,14 +91,12 @@ mod service {
client
.service_builder()
.metadata(metadata.clone())
.api_url(Url::try_from("http://example.com").unwrap())
.start("serviceA", "1.0.0")
.await
.unwrap()
.endpoint_builder()
.name("name")
.metadata(endpoint_metadata.clone())
.schema("request", "response")
.add("products")
.await
.unwrap();
Expand All @@ -120,25 +115,6 @@ mod service {
assert_eq!(metadata, info.metadata);
//TODO: test rest of fields

let schema_reply = client.new_inbox();
let mut schemas = client.subscribe(schema_reply.clone()).await.unwrap();
client
.publish_with_reply("$SRV.SCHEMA".to_string(), schema_reply, "".into())
.await
.unwrap();
let schema = schemas
.next()
.await
.map(|message| serde_json::from_slice::<service::Schema>(&message.payload).unwrap())
.unwrap();
assert_eq!(
schema.endpoints.first().unwrap().schema.clone().unwrap(),
service::endpoint::Schema {
request: "request".to_string(),
response: "response".to_string(),
}
);

let reply = client.new_inbox();
let mut responses = client.subscribe(reply.clone()).await.unwrap();
client
Expand Down Expand Up @@ -167,7 +143,6 @@ mod service {
version: "1.0.0".to_string(),
stats_handler: None,
metadata: None,
api_url: None,
})
.await
.unwrap();
Expand All @@ -179,7 +154,6 @@ mod service {
version: "2.0.0".to_string(),
stats_handler: None,
metadata: None,
api_url: None,
})
.await
.unwrap();
Expand All @@ -206,7 +180,6 @@ mod service {
description: None,
stats_handler: None,
metadata: None,
api_url: None,
})
.await
.unwrap();
Expand Down Expand Up @@ -246,7 +219,6 @@ mod service {
description: None,
stats_handler: None,
metadata: None,
api_url: None,
})
.await
.unwrap();
Expand Down

0 comments on commit d965068

Please sign in to comment.