1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use super::{
    super::{
        super::{Endpoint, EndpointsGetOptions, EndpointsProvider},
        request::SyncInnerRequest,
        ApiResult, InnerRequestParts, RetriedStatsInfo, RetryDecision, SyncRequestBody, SyncResponse,
    },
    error::TryErrorWithExtensions,
    ip_addrs_set::IpAddrsSet,
    try_endpoints::try_endpoints,
};
use log::debug;
use qiniu_http::Extensions;

pub(in super::super) fn request_call<E: EndpointsProvider>(
    request: SyncInnerRequest<'_, E>,
) -> ApiResult<SyncResponse> {
    let (parts, mut body, into_endpoints, service_name, extensions) = request.split();
    let options = EndpointsGetOptions::builder().service_names(service_name).build();
    let endpoints = into_endpoints.get_endpoints(options)?;
    let mut tried_ips = IpAddrsSet::default();
    let mut retried = RetriedStatsInfo::default();

    return match try_preferred_endpoints(
        endpoints.preferred(),
        &parts,
        &mut body,
        extensions,
        &mut tried_ips,
        &mut retried,
    ) {
        Ok(response) => Ok(response),
        Err(err)
            if err.retry_decision() == RetryDecision::TryAlternativeEndpoints
                && !endpoints.alternative().is_empty() =>
        {
            let (_, extensions) = err.split();
            retried.switch_to_alternative_endpoints();
            debug!("Switch to alternative endpoints");
            try_alternative_endpoints(
                endpoints.alternative(),
                &parts,
                &mut body,
                extensions,
                &mut tried_ips,
                &mut retried,
            )
        }
        Err(err) => Err(err.into_response_error()),
    };

    fn try_preferred_endpoints(
        endpoints: &[Endpoint],
        parts: &InnerRequestParts<'_>,
        body: &mut SyncRequestBody<'_>,
        extensions: Extensions,
        tried_ips: &mut IpAddrsSet,
        retried: &mut RetriedStatsInfo,
    ) -> Result<SyncResponse, TryErrorWithExtensions> {
        try_endpoints(endpoints, parts, body, extensions, tried_ips, retried, true)
    }

    fn try_alternative_endpoints(
        endpoints: &[Endpoint],
        parts: &InnerRequestParts<'_>,
        body: &mut SyncRequestBody<'_>,
        extensions: Extensions,
        tried_ips: &mut IpAddrsSet,
        retried: &mut RetriedStatsInfo,
    ) -> ApiResult<SyncResponse> {
        try_endpoints(endpoints, parts, body, extensions, tried_ips, retried, false)
            .map_err(|err| err.into_response_error())
    }
}

#[cfg(feature = "async")]
use super::{
    super::{request::AsyncInnerRequest, AsyncRequestBody, AsyncResponse},
    try_endpoints::async_try_endpoints,
};

#[cfg(feature = "async")]
pub(in super::super) async fn async_request_call<E: EndpointsProvider>(
    request: AsyncInnerRequest<'_, E>,
) -> ApiResult<AsyncResponse> {
    let (parts, mut body, into_endpoints, service_name, extensions) = request.split();
    let options = EndpointsGetOptions::builder().service_names(service_name).build();
    let endpoints = into_endpoints.async_get_endpoints(options).await?;
    let mut tried_ips = IpAddrsSet::default();
    let mut retried = RetriedStatsInfo::default();

    return match try_preferred_endpoints(
        endpoints.preferred(),
        &parts,
        &mut body,
        extensions,
        &mut tried_ips,
        &mut retried,
    )
    .await
    {
        Ok(response) => Ok(response),
        Err(err)
            if err.retry_decision() == RetryDecision::TryAlternativeEndpoints
                && !endpoints.alternative().is_empty() =>
        {
            let (_, extensions) = err.split();
            retried.switch_to_alternative_endpoints();
            debug!("Switch to alternative endpoints");
            try_alternative_endpoints(
                endpoints.alternative(),
                &parts,
                &mut body,
                extensions,
                &mut tried_ips,
                &mut retried,
            )
            .await
        }
        Err(err) => Err(err.into_response_error()),
    };

    async fn try_preferred_endpoints(
        endpoints: &[Endpoint],
        parts: &InnerRequestParts<'_>,
        body: &mut AsyncRequestBody<'_>,
        extensions: Extensions,
        tried_ips: &mut IpAddrsSet,
        retried: &mut RetriedStatsInfo,
    ) -> Result<AsyncResponse, TryErrorWithExtensions> {
        async_try_endpoints(endpoints, parts, body, extensions, tried_ips, retried, true).await
    }

    async fn try_alternative_endpoints(
        endpoints: &[Endpoint],
        parts: &InnerRequestParts<'_>,
        body: &mut AsyncRequestBody<'_>,
        extensions: Extensions,
        tried_ips: &mut IpAddrsSet,
        retried: &mut RetriedStatsInfo,
    ) -> ApiResult<AsyncResponse> {
        async_try_endpoints(endpoints, parts, body, extensions, tried_ips, retried, false)
            .await
            .map_err(|err| err.into_response_error())
    }
}