Make endpoint resolution async (#2816)

## Motivation and Context
Make endpoint resolution async from the perspective of the orchestrator.
- #2608 

## Description
Change the endpoint trait in the orchestrator to be asynchronous

## Testing
- compiled code

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

Author: rcoh

rust-runtime/aws-smithy-runtime-api/src/client/orchestrator.rs

@@ -135,7 +135,7 @@ impl Storable for EndpointResolverParams {
 }
 
 pub trait EndpointResolver: Send + Sync + fmt::Debug {
-    fn resolve_endpoint(&self, params: &EndpointResolverParams) -> Result<Endpoint, BoxError>;
+    fn resolve_endpoint(&self, params: &EndpointResolverParams) -> Future<Endpoint>;
 }
 
 #[derive(Debug)]
@@ -148,7 +148,7 @@ impl DynEndpointResolver {
 }
 
 impl EndpointResolver for DynEndpointResolver {
-    fn resolve_endpoint(&self, params: &EndpointResolverParams) -> Result<Endpoint, BoxError> {
+    fn resolve_endpoint(&self, params: &EndpointResolverParams) -> Future<Endpoint> {
         self.0.resolve_endpoint(params)
     }
 }

rust-runtime/aws-smithy-runtime/src/client/orchestrator.rs

@@ -261,7 +261,7 @@ async fn try_attempt(
     stop_point: StopPoint,
 ) {
     halt_on_err!([ctx] => interceptors.read_before_attempt(ctx, cfg));
-    halt_on_err!([ctx] => orchestrate_endpoint(ctx, cfg).map_err(OrchestratorError::other));
+    halt_on_err!([ctx] => orchestrate_endpoint(ctx, cfg).await.map_err(OrchestratorError::other));
     halt_on_err!([ctx] => interceptors.modify_before_signing(ctx, cfg));
     halt_on_err!([ctx] => interceptors.read_before_signing(ctx, cfg));
 

rust-runtime/aws-smithy-runtime/src/client/orchestrator/endpoints.rs

@@ -12,7 +12,7 @@ use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::config_bag_accessors::ConfigBagAccessors;
 use aws_smithy_runtime_api::client::interceptors::context::InterceptorContext;
 use aws_smithy_runtime_api::client::orchestrator::{
-    EndpointResolver, EndpointResolverParams, HttpRequest,
+    EndpointResolver, EndpointResolverParams, Future, HttpRequest,
 };
 use aws_smithy_types::config_bag::{ConfigBag, Storable, StoreReplace};
 use aws_smithy_types::endpoint::Endpoint;
@@ -40,8 +40,10 @@ impl StaticUriEndpointResolver {
 }
 
 impl EndpointResolver for StaticUriEndpointResolver {
-    fn resolve_endpoint(&self, _params: &EndpointResolverParams) -> Result<Endpoint, BoxError> {
-        Ok(Endpoint::builder().url(self.endpoint.to_string()).build())
+    fn resolve_endpoint(&self, _params: &EndpointResolverParams) -> Future<Endpoint> {
+        Future::ready(Ok(Endpoint::builder()
+            .url(self.endpoint.to_string())
+            .build()))
     }
 }
 
@@ -86,17 +88,19 @@ impl<Params> EndpointResolver for DefaultEndpointResolver<Params>
 where
     Params: Debug + Send + Sync + 'static,
 {
-    fn resolve_endpoint(&self, params: &EndpointResolverParams) -> Result<Endpoint, BoxError> {
-        match params.get::<Params>() {
-            Some(params) => Ok(self.inner.resolve_endpoint(params)?),
+    fn resolve_endpoint(&self, params: &EndpointResolverParams) -> Future<Endpoint> {
+        let ep = match params.get::<Params>() {
+            Some(params) => self.inner.resolve_endpoint(params).map_err(Box::new),
             None => Err(Box::new(ResolveEndpointError::message(
                 "params of expected type was not present",
             ))),
         }
+        .map_err(|e| e as _);
+        Future::ready(ep)
     }
 }
 
-pub(super) fn orchestrate_endpoint(
+pub(super) async fn orchestrate_endpoint(
     ctx: &mut InterceptorContext,
     cfg: &mut ConfigBag,
 ) -> Result<(), BoxError> {
@@ -105,7 +109,7 @@ pub(super) fn orchestrate_endpoint(
     let request = ctx.request_mut().expect("set during serialization");
 
     let endpoint_resolver = cfg.endpoint_resolver();
-    let endpoint = endpoint_resolver.resolve_endpoint(params)?;
+    let endpoint = endpoint_resolver.resolve_endpoint(params).await?;
     apply_endpoint(request, &endpoint, endpoint_prefix)?;
 
     // Make the endpoint config available to interceptors