Skip to content

Commit

Permalink
Use <service>::endpoint::ResolveEndpoint and delete aws_smithy_http…
Browse files Browse the repository at this point in the history 
…::Endpoint
  • Loading branch information
@rcoh
rcoh committed Oct 18, 2023
1 parent bcfc211 commit 3ebc0ad
Show file tree
Hide file tree
Showing 20 changed files with 203 additions and 456 deletions.
12 changes: 12 additions & 0 deletions CHANGELOG.next.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -414,3 +414,15 @@ message = "The `idempotency_provider` field has been removed from config as a pu
references = ["smithy-rs#3072"]
meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
author = "rcoh"

[[smithy-rs]]
message = "The `config::Builder::endpoint_resolver` method no longer accepts `&'static str`. Use `config::Builder::endpoint_url` instead."
references = ["smithy-rs#3078"]
meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
author = "rcoh"

[[smithy-rs]]
message = "**This change has [detailed upgrade guidance](https://github.com/awslabs/smithy-rs/discussions/3079).** <br><br>The endpoint interfaces from `aws-smithy-http` have been removed. Service-specific endpoint resolver traits have been added."
references = ["smithy-rs#3043", "smithy-rs#3078"]
meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
author = "rcoh"
59 changes: 34 additions & 25 deletions aws/rust-runtime/aws-inlineable/src/endpoint_discovery.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,10 @@
use aws_smithy_async::future::BoxFuture;
use aws_smithy_async::rt::sleep::{AsyncSleep, SharedAsyncSleep};
use aws_smithy_async::time::SharedTimeSource;
use aws_smithy_http::endpoint::{ResolveEndpoint, ResolveEndpointError};
use aws_smithy_runtime_api::box_error::BoxError;
use aws_smithy_runtime_api::client::endpoint::{
EndpointFuture, EndpointResolverParams, ResolveEndpoint,
};
use aws_smithy_types::endpoint::Endpoint;
use std::fmt::{Debug, Formatter};
use std::future::Future;
Expand All @@ -20,11 +23,9 @@ use tokio::sync::oneshot::{Receiver, Sender};
/// Endpoint reloader
#[must_use]
pub struct ReloadEndpoint {
loader: Box<
dyn Fn() -> BoxFuture<'static, (Endpoint, SystemTime), ResolveEndpointError> + Send + Sync,
>,
loader: Box<dyn Fn() -> BoxFuture<'static, (Endpoint, SystemTime), BoxError> + Send + Sync>,
endpoint: Arc<Mutex<Option<ExpiringEndpoint>>>,
error: Arc<Mutex<Option<ResolveEndpointError>>>,
error: Arc<Mutex<Option<BoxError>>>,
rx: Receiver<()>,
sleep: SharedAsyncSleep,
time: SharedTimeSource,
Expand Down Expand Up @@ -79,14 +80,14 @@ impl ReloadEndpoint {

#[derive(Debug, Clone)]
pub(crate) struct EndpointCache {
error: Arc<Mutex<Option<ResolveEndpointError>>>,
error: Arc<Mutex<Option<BoxError>>>,
endpoint: Arc<Mutex<Option<ExpiringEndpoint>>>,
// When the sender is dropped, this allows the reload loop to stop
_drop_guard: Arc<Sender<()>>,
}

impl<T> ResolveEndpoint<T> for EndpointCache {
fn resolve_endpoint(&self, _params: &T) -> aws_smithy_http::endpoint::Result {
impl ResolveEndpoint for EndpointCache {
fn resolve_endpoint<'a>(&'a self, _params: &'a EndpointResolverParams) -> EndpointFuture<'a> {
self.resolve_endpoint()
}
}
Expand All @@ -111,9 +112,9 @@ pub(crate) async fn create_cache<F>(
loader_fn: impl Fn() -> F + Send + Sync + 'static,
sleep: SharedAsyncSleep,
time: SharedTimeSource,
) -> Result<(EndpointCache, ReloadEndpoint), ResolveEndpointError>
) -> Result<(EndpointCache, ReloadEndpoint), BoxError>
where
F: Future<Output = Result<(Endpoint, SystemTime), ResolveEndpointError>> + Send + 'static,
F: Future<Output = Result<(Endpoint, SystemTime), BoxError>> + Send + 'static,
{
let error_holder = Arc::new(Mutex::new(None));
let endpoint_holder = Arc::new(Mutex::new(None));
Expand All @@ -135,25 +136,24 @@ where
reloader.reload_once().await;
// if we didn't successfully get an endpoint, bail out so the client knows
// configuration failed to work
cache.resolve_endpoint()?;
cache.resolve_endpoint().await?;
Ok((cache, reloader))
}

impl EndpointCache {
fn resolve_endpoint(&self) -> aws_smithy_http::endpoint::Result {
fn resolve_endpoint(&self) -> EndpointFuture<'_> {
tracing::trace!("resolving endpoint from endpoint discovery cache");
self.endpoint
let ep = self
.endpoint
.lock()
.unwrap()
.as_ref()
.map(|e| e.endpoint.clone())
.ok_or_else(|| {
self.error
.lock()
.unwrap()
.take()
.unwrap_or_else(|| ResolveEndpointError::message("no endpoint loaded"))
})
let error: Option<BoxError> = self.error.lock().unwrap().take();
error.unwrap_or_else(|| "err".into())
});
EndpointFuture::ready(ep)
}
}

Expand Down Expand Up @@ -215,21 +215,21 @@ mod test {
.await
.expect("returns an endpoint");
assert_eq!(
cache.resolve_endpoint().expect("ok").url(),
cache.resolve_endpoint().await.expect("ok").url(),
"http://foo.com/1"
);
// 120 second buffer
reloader
.reload_increment(expiry - Duration::from_secs(240))
.await;
assert_eq!(
cache.resolve_endpoint().expect("ok").url(),
cache.resolve_endpoint().await.expect("ok").url(),
"http://foo.com/1"
);

reloader.reload_increment(expiry).await;
assert_eq!(
cache.resolve_endpoint().expect("ok").url(),
cache.resolve_endpoint().await.expect("ok").url(),
"http://foo.com/2"
);
}
Expand Down Expand Up @@ -266,18 +266,27 @@ mod test {
gate.expect_sleep().await.duration(),
Duration::from_secs(60)
);
assert_eq!(cache.resolve_endpoint().unwrap().url(), "http://foo.com/1");
assert_eq!(
cache.resolve_endpoint().await.unwrap().url(),
"http://foo.com/1"
);
// t = 60

let sleep = gate.expect_sleep().await;
// we're still holding the drop guard, so we haven't expired yet.
assert_eq!(cache.resolve_endpoint().unwrap().url(), "http://foo.com/1");
assert_eq!(
cache.resolve_endpoint().await.unwrap().url(),
"http://foo.com/1"
);
assert_eq!(sleep.duration(), Duration::from_secs(60));
sleep.allow_progress();
// t = 120

let sleep = gate.expect_sleep().await;
assert_eq!(cache.resolve_endpoint().unwrap().url(), "http://foo.com/2");
assert_eq!(
cache.resolve_endpoint().await.unwrap().url(),
"http://foo.com/2"
);
sleep.allow_progress();

let sleep = gate.expect_sleep().await;
Expand Down
22 changes: 10 additions & 12 deletions ...rc/main/kotlin/software/amazon/smithy/rustsdk/customize/timestream/TimestreamDecorator.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -55,11 +55,9 @@ class TimestreamDecorator : ClientCodegenDecorator {
// helper function to resolve an endpoint given a base client
rustTemplate(
"""
async fn resolve_endpoint(client: &crate::Client) -> Result<(#{Endpoint}, #{SystemTime}), #{ResolveEndpointError}> {
async fn resolve_endpoint(client: &crate::Client) -> Result<(#{Endpoint}, #{SystemTime}), #{BoxError}> {
let describe_endpoints =
client.describe_endpoints().send().await.map_err(|e| {
#{ResolveEndpointError}::from_source("failed to call describe_endpoints", e)
})?;
client.describe_endpoints().send().await?;
let endpoint = describe_endpoints.endpoints().get(0).unwrap();
let expiry = client.config().time_source().expect("checked when ep discovery was enabled").now()
+ #{Duration}::from_secs(endpoint.cache_period_in_minutes() as u64 * 60);
Expand All @@ -75,7 +73,7 @@ class TimestreamDecorator : ClientCodegenDecorator {
/// Enable endpoint discovery for this client
///
/// This method MUST be called to construct a working client.
pub async fn with_endpoint_discovery_enabled(self) -> #{Result}<(Self, #{endpoint_discovery}::ReloadEndpoint), #{ResolveEndpointError}> {
pub async fn with_endpoint_discovery_enabled(self) -> #{Result}<(Self, #{endpoint_discovery}::ReloadEndpoint), #{BoxError}> {
let handle = self.handle.clone();
// The original client without endpoint discover gets moved into the endpoint discovery
Expand All @@ -92,22 +90,22 @@ class TimestreamDecorator : ClientCodegenDecorator {
.expect("endpoint discovery requires the client config to have a time source"),
).await?;
let client_with_discovery = crate::Client::from_conf(
handle.conf.to_builder()
.endpoint_resolver(#{SharedEndpointResolver}::new(resolver))
.build()
);
use #{IntoShared};
let mut conf = handle.conf.to_builder();
conf.set_endpoint_resolver(Some(resolver.into_shared()));
let client_with_discovery = crate::Client::from_conf(conf.build());
Ok((client_with_discovery, reloader))
}
}
""",
*RuntimeType.preludeScope,
"Arc" to RuntimeType.Arc,
"Duration" to RuntimeType.std.resolve("time::Duration"),
"SharedEndpointResolver" to RuntimeType.smithyHttp(codegenContext.runtimeConfig)
.resolve("endpoint::SharedEndpointResolver"),
"SystemTime" to RuntimeType.std.resolve("time::SystemTime"),
"endpoint_discovery" to endpointDiscovery.toType(),
"BoxError" to RuntimeType.boxError(codegenContext.runtimeConfig),
"IntoShared" to RuntimeType.smithyRuntimeApi(codegenContext.runtimeConfig).resolve("shared::IntoShared"),
*Types(codegenContext.runtimeConfig).toArray(),
)
}
Expand Down
2 changes: 1 addition & 1 deletion ...ent/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/ClientRustModule.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ class ClientModuleDocProvider(
ClientRustModule.Config.endpoint -> strDoc("Types needed to configure endpoint resolution.")
ClientRustModule.Config.retry -> strDoc("Retry configuration.")
ClientRustModule.Config.timeout -> strDoc("Timeout configuration.")
ClientRustModule.Config.interceptors -> strDoc("Types needed to implement [`Interceptor`](crate::config::Interceptor).")
ClientRustModule.Config.interceptors -> strDoc("Types needed to implement [`Intercept`](crate::config::Intercept).")
ClientRustModule.Error -> strDoc("Common errors and error handling utilities.")
ClientRustModule.Operation -> strDoc("All operations that this crate can perform.")
ClientRustModule.Meta -> strDoc("Information about this crate.")
Expand Down
83 changes: 34 additions & 49 deletions ...software/amazon/smithy/rust/codegen/client/smithy/endpoint/EndpointConfigCustomization.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,10 @@ package software.amazon.smithy.rust.codegen.client.smithy.endpoint

import software.amazon.smithy.rust.codegen.client.smithy.ClientCodegenContext
import software.amazon.smithy.rust.codegen.client.smithy.ClientRustModule
import software.amazon.smithy.rust.codegen.client.smithy.endpoint.generators.serviceSpecificResolver
import software.amazon.smithy.rust.codegen.client.smithy.generators.config.ConfigCustomization
import software.amazon.smithy.rust.codegen.client.smithy.generators.config.ServiceConfig
import software.amazon.smithy.rust.codegen.core.rustlang.Writable
import software.amazon.smithy.rust.codegen.core.rustlang.rust
import software.amazon.smithy.rust.codegen.core.rustlang.rustTemplate
import software.amazon.smithy.rust.codegen.core.rustlang.writable
import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType
Expand All @@ -26,23 +26,21 @@ internal class EndpointConfigCustomization(
ConfigCustomization() {
private val runtimeConfig = codegenContext.runtimeConfig
private val moduleUseName = codegenContext.moduleUseName()
private val types = Types(runtimeConfig)
private val epModule = RuntimeType.smithyRuntimeApi(runtimeConfig).resolve("client::endpoint")
private val epRuntimeModule = RuntimeType.smithyRuntime(runtimeConfig).resolve("client::orchestrator::endpoints")

private val codegenScope = arrayOf(
*preludeScope,
"DefaultEndpointResolver" to RuntimeType.smithyRuntime(runtimeConfig).resolve("client::orchestrator::endpoints::DefaultEndpointResolver"),
"Endpoint" to RuntimeType.smithyHttp(runtimeConfig).resolve("endpoint::Endpoint"),
"OldSharedEndpointResolver" to types.sharedEndpointResolver,
"Params" to typesGenerator.paramsStruct(),
"IntoShared" to RuntimeType.smithyRuntimeApi(runtimeConfig).resolve("shared::IntoShared"),
"Resolver" to RuntimeType.smithyRuntime(runtimeConfig).resolve("client::config_override::Resolver"),
"SharedEndpointResolver" to RuntimeType.smithyRuntimeApi(runtimeConfig).resolve("client::endpoint::SharedEndpointResolver"),
"SmithyResolver" to types.resolveEndpoint,
"SharedEndpointResolver" to epModule.resolve("SharedEndpointResolver"),
"StaticUriEndpointResolver" to epRuntimeModule.resolve("StaticUriEndpointResolver"),
"ServiceSpecificResolver" to codegenContext.serviceSpecificResolver(),
)

override fun section(section: ServiceConfig): Writable {
return writable {
val sharedEndpointResolver = "#{OldSharedEndpointResolver}<#{Params}>"
val resolverTrait = "#{SmithyResolver}<#{Params}>"
when (section) {
is ServiceConfig.ConfigImpl -> {
rustTemplate(
Expand All @@ -57,44 +55,17 @@ internal class EndpointConfigCustomization(
}

ServiceConfig.BuilderImpl -> {
val endpointModule = ClientRustModule.Config.endpoint.fullyQualifiedPath()
.replace("crate::", "$moduleUseName::")
// if there are no rules, we don't generate a default resolver—we need to also suppress those docs.
val defaultResolverDocs = if (typesGenerator.defaultResolver() != null) {
val endpointModule = ClientRustModule.Config.endpoint.fullyQualifiedPath()
.replace("crate::", "$moduleUseName::")
"""
///
/// When unset, the client will used a generated endpoint resolver based on the endpoint resolution
/// rules for `$moduleUseName`.
///
/// ## Examples
/// ```no_run
/// use aws_smithy_http::endpoint;
/// use $endpointModule::{Params as EndpointParams, DefaultResolver};
/// /// Endpoint resolver which adds a prefix to the generated endpoint
/// ##[derive(Debug)]
/// struct PrefixResolver {
/// base_resolver: DefaultResolver,
/// prefix: String
/// }
/// impl endpoint::ResolveEndpoint<EndpointParams> for PrefixResolver {
/// fn resolve_endpoint(&self, params: &EndpointParams) -> endpoint::Result {
/// self.base_resolver
/// .resolve_endpoint(params)
/// .map(|ep|{
/// let url = ep.url().to_string();
/// ep.into_builder().url(format!("{}.{}", &self.prefix, url)).build()
/// })
/// }
/// }
/// let prefix_resolver = PrefixResolver {
/// base_resolver: DefaultResolver::new(),
/// prefix: "subdomain".to_string()
/// };
/// let config = $moduleUseName::Config::builder().endpoint_resolver(prefix_resolver);
/// ```
"""
} else {
""
"/// This service does not define a default endpoint resolver."
}
if (codegenContext.settings.codegenConfig.includeEndpointUrlConfig) {
rustTemplate(
Expand All @@ -120,9 +91,8 @@ internal class EndpointConfigCustomization(
##[allow(deprecated)]
self.set_endpoint_resolver(
endpoint_url.map(|url| {
#{OldSharedEndpointResolver}::new(
#{Endpoint}::immutable(url).expect("invalid endpoint URL")
)
use #{IntoShared};
#{StaticUriEndpointResolver}::uri(url).into_shared()
})
);
self
Expand All @@ -134,12 +104,28 @@ internal class EndpointConfigCustomization(
rustTemplate(
"""
/// Sets the endpoint resolver to use when making requests.
$defaultResolverDocs
///
/// Note: setting an endpoint resolver will replace any endpoint URL that has been set.
///
$defaultResolverDocs
pub fn endpoint_resolver(mut self, endpoint_resolver: impl $resolverTrait + 'static) -> Self {
self.set_endpoint_resolver(#{Some}(#{OldSharedEndpointResolver}::new(endpoint_resolver)));
/// This method accepts an endpoint resolver [specific to this service](#{ServiceSpecificResolver}). If you want to
/// provide a shared endpoint resolver, use [`Self::set_endpoint_resolver`].
/// ## Examples
/// ```no_run
/// use $endpointModule::{ResolveEndpoint, EndpointFuture, Params, Endpoint};
/// ##[derive(Debug)]
/// struct StageResolver { stage: String }
/// impl ResolveEndpoint for StageResolver {
/// fn resolve_endpoint(&self, params: &Params) -> EndpointFuture<'_> {
/// let stage = &self.stage;
/// EndpointFuture::ready(Ok(Endpoint::builder().url(format!("{stage}.myservice.com")).build()))
/// }
/// }
/// let resolver = StageResolver { stage: std::env::var("STAGE").unwrap() };
/// let config = $moduleUseName::Config::builder().endpoint_resolver(resolver).build();
/// let client = $moduleUseName::Client::from_conf(config);
/// ```
pub fn endpoint_resolver(mut self, endpoint_resolver: impl #{ServiceSpecificResolver} + 'static) -> Self {
self.set_endpoint_resolver(#{Some}(endpoint_resolver.into_shared_resolver()));
self
}
Expand All @@ -153,13 +139,12 @@ internal class EndpointConfigCustomization(

rustTemplate(
"""
pub fn set_endpoint_resolver(&mut self, endpoint_resolver: #{Option}<$sharedEndpointResolver>) -> &mut Self {
self.runtime_components.set_endpoint_resolver(endpoint_resolver.map(|r|#{wrap_resolver}));
pub fn set_endpoint_resolver(&mut self, endpoint_resolver: #{Option}<#{SharedEndpointResolver}>) -> &mut Self {
self.runtime_components.set_endpoint_resolver(endpoint_resolver);
self
}
""",
*codegenScope,
"wrap_resolver" to codegenContext.wrapResolver { rust("r") },
)
}

Expand Down
Loading

0 comments on commit 3ebc0ad

Please sign in to comment.