gapirs_common/
api.rs

1use super::*;
2use crate::errors::ApiCallError::Call;
3use crate::errors::CallError;
4use crate::errors::GetUrlError;
5use crate::errors::NonSuccessStatusCodeError;
6use crate::errors::StdResult;
7use crate::errors::{ApiCallError, ResumableUploadError};
8use crate::media::resumable::{LimitedStream, PendingChunks};
9use crate::media::{AsyncMediaUpload, MediaDownloadResponseStream};
10use crate::utils::vec_helper::SortTupleIteratorBySortKeys;
11pub use client::media;
12pub use client::media::resumable::ResumableBody;
13pub use client::media::resumable::ResumableMediaUpload;
14use client::media::resumable::ResumableState;
15pub use client::media::AsyncMediaUploadStream;
16pub use client::media::MediaDownload;
17pub use client::media::MediaDownloadStream;
18pub use client::media::MediaUpload;
19pub use client::media::MediaUploadProtocol;
20pub use client::media::MediaUploadStream;
21pub use client::OutgoingBodyContent;
22use core::fmt::Debug;
23use futures_util::lock::Mutex;
24use futures_util::{AsyncSeekExt, StreamExt};
25use http_body_util::{BodyExt, Full, StreamBody};
26use hyper::body::Frame;
27use hyper::body::{Bytes, Incoming};
28use hyper::header::CONTENT_LENGTH;
29use hyper::header::CONTENT_RANGE;
30use hyper::header::CONTENT_TYPE;
31use hyper::header::{AUTHORIZATION, RANGE};
32use hyper::Request;
33use hyper::Response;
34use hyper::{Method, StatusCode};
35use hyper_util::client::legacy::connect::Connect;
36use hyper_util::client::legacy::Client;
37use serde::de::DeserializeOwned;
38use serde::Deserialize;
39use std::collections::HashMap;
40use std::io::{SeekFrom, Write};
41use std::pin::Pin;
42use std::str::FromStr;
43use std::task::Context;
44use std::task::Poll;
45use url::Url;
46
47mod client;
48
49type Result<T> = StdResult<T, ApiCallError>;
50
51async fn call_no_parse_response<'a, H: Hub, R2, A: ApiCall<'a, R2, H>>(
52    call: &mut A,
53) -> Result<Response<Incoming>> {
54    let mut url: Url = call.get_url()?; //has to be called, before any call.take_*()-function from since it would change the url
55    let media_upload = call.take_media_upload();
56    let outgoing_body = call.take_body();
57    let resumable_media = call.take_resumable_media_upload();
58    let incoming_body = call.get_media_download().and_then(|x| x.get_range());
59    let hub = call.get_hub()?;
60    let mut outgoing_body = OutgoingBodyContent::from_body_and_media(outgoing_body, media_upload);
61
62    let is_resuming_upload;
63    if let Some(resumable) = &resumable_media {
64        match resumable.state {
65            ResumableState::Resuming => {
66                if let Some(resume_url) = &resumable.url {
67                    url = resume_url.clone();
68                    is_resuming_upload = true;
69                    outgoing_body = OutgoingBodyContent::Empty; //TODO: check if this is always empty when resuming (it should be right?)
70                } else {
71                    is_resuming_upload = false;
72                }
73            }
74            _ => {
75                is_resuming_upload = false;
76            }
77        }
78    } else {
79        is_resuming_upload = false;
80    }
81
82    let url = url.as_str();
83    println!("url: '{}'", url);
84    let body_len = outgoing_body.get_length();
85    dbg!(&outgoing_body, &body_len, url);
86    let mut req = hyper::Request::builder()
87        .method(A::get_request_method())
88        .uri(url)
89        .header(CONTENT_LENGTH, body_len);
90
91    if let Some(range) = incoming_body {
92        dbg!("Setting range header", &range);
93        req = req.header(RANGE, format!("bytes={}", range));
94    }
95    if is_resuming_upload {
96        let total_range = resumable_media
97            .as_ref()
98            .and_then(|x| x.media_body.as_ref())
99            .and_then(|x| x.length)
100            .map(|x| x.to_string())
101            .unwrap_or("*".to_string());
102        req = req.header(CONTENT_RANGE, format!("bytes */{}", total_range));
103    }
104    if let Some(content_type) = outgoing_body.get_content_type() {
105        req = req.header(CONTENT_TYPE, content_type);
106    }
107    dbg!(&req);
108    let scopes = A::get_scopes();
109    if !scopes.is_empty() {
110        req = req.header(
111            AUTHORIZATION,
112            format!("Bearer {}", hub.get_token(&scopes).await?),
113        );
114    }
115
116    let req = req.body(outgoing_body).map_err(CallError::InvalidRequest)?;
117    // dbg!(&req);
118    let client = hub.get_client();
119    let mut response = client.request(req).await.map_err(CallError::Request)?;
120
121    // dbg!(&response);
122    let status_code = response.status();
123    dbg!(&status_code);
124    let resume_incompleted_upload = is_resuming_upload && status_code == 308;
125    if status_code.is_success() || resume_incompleted_upload {
126        if let Some(mut resumable) = resumable_media {
127            dbg!("Resumable upload", &resumable, &response);
128            if is_resuming_upload {
129                handle_resumable_upload_resume_response(&response, &mut resumable)?;
130            } else {
131                handle_resumable_upload_init_response(&response, &mut resumable)?;
132            }
133
134            while let ResumableState::Sending(_) = &resumable.state {
135                dbg!("Looping through resumable upload");
136                response = resumable_upload_chunk(&mut resumable, hub).await?;
137            }
138            dbg!(&resumable, &response);
139        }
140
141        Ok(response)
142        // let body = response.into_body();
143        // Ok(body)
144    } else {
145        let body = response.into_body();
146        let body = body.collect().await.map_err(CallError::GetBody)?.to_bytes();
147        dbg!(&body);
148        Err(CallError::NonSuccessStatusCode(
149            NonSuccessStatusCodeError::from((
150                status_code,
151                String::from_utf8_lossy(&body).to_string(),
152            )),
153        ))?
154    }
155}
156
157async fn call<'a, R, H: Hub, R2, A: ApiCall<'a, R2, H>>(call: &mut A) -> Result<R>
158where
159    R: Debug + DeserializeOwned,
160{
161    if call.is_media_download() {
162        return Err(CallError::DownloadSetInNotDownloadCall.into());
163    }
164    let body = call_no_parse_response(call).await?;
165    let body = body
166        .into_body()
167        .collect()
168        .await
169        .map_err(CallError::GetBody)?
170        .to_bytes();
171    dbg!(&body);
172    let response: R = serde_json::from_slice(&body).map_err(CallError::ParseResponse)?;
173    Ok(response)
174}
175async fn call_download<'a, H: Hub, R2, A: ApiCall<'a, R2, H>>(call: &mut A) -> Result<()> {
176    if !call.is_media_download() {
177        return Err(CallError::NoDownloadTarget.into());
178    }
179    let mut download_stream = call_get_download_response_stream(call).await?;
180
181    let Some(mut writer) = call.take_media_download() else {
182        return Err(CallError::NoDownloadTarget.into());
183    };
184    while let Some(frame) = download_stream.stream.next().await {
185        let frame_data = match frame {
186            Err(e) => {
187                return Err(CallError::GetBody(e).into());
188            }
189            Ok(frame_data) => frame_data,
190        };
191        writer
192            .stream
193            .write_all(&frame_data)
194            .map_err(CallError::WriteBodyToStream)?;
195    }
196    Ok(())
197}
198
199async fn call_get_download_response_stream<'a, H: Hub, R2, A: ApiCall<'a, R2, H>>(
200    call: &mut A,
201) -> Result<MediaDownloadResponseStream> {
202    let response = call_no_parse_response(call).await?;
203    let content_headers = response.headers().to_owned();
204    let body = response.into_body();
205    let stream = body.into_data_stream();
206    let download_stream = MediaDownloadResponseStream {
207        stream,
208        headers: content_headers,
209    };
210    Ok(download_stream)
211}
212
213async fn resumable_upload_chunk(
214    resumable: &mut ResumableBody,
215    hub: &impl Hub,
216) -> Result<Response<Incoming>> {
217    println!("Uploading chunk");
218    let single_chunk = resumable.is_single_chunk_upload().unwrap_or(true);
219    let url = resumable.url.as_ref().ok_or(ResumableUploadError::NoUrl)?;
220
221    let media = resumable
222        .media_body
223        .as_mut()
224        .ok_or(ResumableUploadError::NoMediaBody)?;
225    let total_length = media.length;
226    if let ResumableState::Sending(ranges) = &resumable.state {
227        if let Some(range) = ranges.ranges.first() {
228            let range = range.resolve_range(total_length)?;
229            println!("Chunk range {}/{}", range.start, range.end);
230            {
231                let mut stream = media.body.lock().await;
232                stream
233                    .seek(SeekFrom::Start(range.start))
234                    .await
235                    .map_err(ResumableUploadError::SeekPosition)?;
236            }
237            let body_len = range.end - range.start + 1; //ranges are inclusive
238            let mut req_builder = hyper::Request::builder()
239                .method(Method::PUT)
240                .uri(url.as_str())
241                .header(CONTENT_LENGTH, body_len);
242
243            req_builder = req_builder.header(CONTENT_TYPE, &media.mime_type);
244            let chunk_length = if single_chunk {
245                total_length.unwrap()
246            } else {
247                body_len
248            };
249            dbg!(&chunk_length);
250            let stream = LimitedStream::new(media.body.clone(), chunk_length)
251                .await
252                .unwrap();
253            let body = OutgoingBodyContent::AsyncStream(AsyncMediaUpload::new(
254                &media.mime_type,
255                stream,
256                chunk_length,
257            ));
258
259            let req = req_builder.body(body).unwrap();
260            let client = hub.get_client();
261            dbg!("sending request");
262            let response = client.request(req).await.map_err(CallError::Request)?;
263            dbg!(&response);
264            handle_resumable_upload_resume_response(&response, resumable)?;
265            dbg!("done with chunk");
266            return Ok(response);
267        } else {
268            dbg!("missing ranges");
269        }
270    } else {
271        dbg!("wrong state", &resumable.state);
272    }
273
274    todo!()
275}
276
277fn handle_resumable_upload_init_response(
278    response: &Response<Incoming>,
279    resumable: &mut ResumableBody,
280) -> StdResult<(), ResumableUploadError> {
281    let x = response
282        .headers()
283        .get("Location")
284        .ok_or(ResumableUploadError::NoOrInvalidLocationHeader)?
285        .to_str()
286        .map_err(|_| ResumableUploadError::NoOrInvalidLocationHeader)?;
287    let x = Url::from_str(x)?;
288    resumable.url = Some(x.clone());
289    resumable.call_save_url(x);
290    resumable.state = ResumableState::Sending(PendingChunks::full(
291        resumable
292            .media_body
293            .as_ref()
294            .ok_or(ResumableUploadError::NoMediaBody)?
295            .length
296            .ok_or(ResumableUploadError::NoMediaBody)?,
297    ));
298    Ok(())
299}
300
301fn handle_resumable_upload_resume_response(
302    response: &Response<Incoming>,
303    resumable: &mut ResumableBody,
304) -> StdResult<(), ResumableUploadError> {
305    let status = response.status();
306    if status.is_success() {
307        resumable.state = ResumableState::Done;
308    } else {
309        let x = response
310            .headers()
311            .get("Range")
312            .map(|x| x.to_str().unwrap_or_default().to_string());
313        let range = if let Some(x) = x {
314            PendingChunks::from_ranges(x)?
315        } else {
316            PendingChunks::full(
317                resumable
318                    .media_body
319                    .as_ref()
320                    .ok_or(ResumableUploadError::NoMediaBody)?
321                    .length
322                    .ok_or(ResumableUploadError::NoMediaBody)?,
323            )
324        };
325        resumable.state = ResumableState::Sending(range);
326    }
327    Ok(())
328}
329
330pub trait ApiCall<'a, R, H: Hub>: ApiCallBase<'a, H, R> + Sized {
331    fn call(&mut self) -> impl Future<Output = Result<R>>
332    where
333        R: Debug + DeserializeOwned,
334    {
335        call(self)
336    }
337    fn call_custom<R2>(&mut self) -> impl Future<Output = Result<R2>>
338    where
339        R2: Debug + DeserializeOwned,
340    {
341        call(self)
342    }
343    fn call_download(&mut self) -> impl Future<Output = Result<()>> {
344        call_download(self)
345    }
346    fn get_url(&self) -> Result<Url> {
347        let hub = self.get_hub()?;
348        let base_url = &hub.get_base_url();
349        let url = match self.get_protocol()? {
350            MediaUploadProtocol::None => {
351                let service_path = hub.get_service_path();
352                self.get_normal_url(&format!("{}{}", base_url, service_path))?
353            }
354            MediaUploadProtocol::Simple => self.get_url_media_upload_simple(base_url)?,
355            MediaUploadProtocol::Resumable => self.get_url_media_upload_resumable(base_url)?,
356        };
357        Ok(url)
358    }
359    fn get_protocol(&self) -> Result<MediaUploadProtocol> {
360        Ok(
361            match (self.has_media_upload(), self.has_resumable_media_upload()) {
362                (true, false) => MediaUploadProtocol::Simple,
363                (false, true) => MediaUploadProtocol::Resumable,
364                (false, false) => MediaUploadProtocol::None,
365                (true, true) => return Err(ApiCallError::MultipleMediaUploads),
366            },
367        )
368    }
369    fn get_url_media_upload_simple(&self, base_url: &str) -> StdResult<Url, GetUrlError> {
370        let path = self
371            .get_media_upload_path_simple()
372            .ok_or(GetUrlError::NoMediaPath)?;
373        self.get_url_with_path(base_url, path, Some("multipart"), false)
374    }
375    fn get_url_media_upload_resumable(&self, base_url: &str) -> StdResult<Url, GetUrlError> {
376        let path = self
377            .get_media_upload_path_resumable()
378            .ok_or(GetUrlError::NoMediaPath)?;
379        self.get_url_with_path(base_url, path, Some("resumable"), false)
380    }
381    fn get_normal_url(&self, base_url: &str) -> StdResult<Url, GetUrlError> {
382        let path = self.get_method_path();
383        let is_download = self.is_media_download();
384        self.get_url_with_path(base_url, path, None, is_download)
385    }
386
387    fn get_url_with_path(
388        &self,
389        base_url: &str,
390        path: impl Into<String>,
391        upload_type_param: Option<&'static str>,
392        download: bool,
393    ) -> StdResult<Url, GetUrlError> {
394        let mut path = path.into();
395        let mut query_params = self.get_query_params();
396        if download {
397            query_params.push(("alt", "media".to_string()));
398        }
399        if let Some(upload_type_param) = upload_type_param {
400            query_params.push(("uploadType", upload_type_param.to_string()));
401        }
402        let query_params = query_params;
403        let mut remaining_parameters = vec![];
404        //replace path parameters
405        for (name, value) in query_params.into_iter() {
406            if path.contains(&format!("{{{}}}", &name)) {
407                path = path.replace(
408                    &format!("{{{}}}", name),
409                    &url::form_urlencoded::byte_serialize(value.as_bytes()).collect::<String>(),
410                );
411            } else {
412                remaining_parameters.push((name, value));
413            }
414        }
415
416        let query_param_order = self.get_query_param_order();
417        let remaining_parameters = remaining_parameters.sort_by_other(&query_param_order);
418
419        let path = path.trim_start_matches('/').trim_end_matches('/');
420        let base_url = base_url.trim_end_matches('/');
421        dbg!(&path, &base_url);
422        let url = format!("{}/{}", base_url, path);
423        let url = Url::parse_with_params(&url, remaining_parameters)?;
424
425        Ok(url)
426    }
427
428    fn with_hub(&mut self, hub: &'a H) -> &mut Self {
429        ApiCallBase::set_hub(self, hub);
430        self
431    }
432}
433impl<'a, T, R: DeserializeOwned + Debug, H: Hub> ApiCall<'a, R, H> for T where
434    T: ApiCallBase<'a, H, R>
435{
436}
437pub trait ApiCallBase<'a, HUB, RETURN> {
438    fn get_method_path(&self) -> &str;
439
440    fn get_media_upload_path_simple(&self) -> Option<&str>;
441    fn get_media_upload_path_resumable(&self) -> Option<&str>;
442
443    fn get_query_params(&self) -> Vec<(&str, String)>;
444    fn get_query_param_order(&self) -> Vec<&str>;
445    fn get_scopes() -> Vec<impl Scope>;
446    fn get_hub(&self) -> Result<&impl Hub>;
447    fn set_hub<'b>(&mut self, hub: &'b HUB)
448    where
449        'b: 'a;
450
451    fn has_body(&self) -> bool;
452    fn take_body(&mut self) -> Option<String>;
453    fn has_media_upload(&self) -> bool;
454    fn is_media_download(&self) -> bool;
455    fn take_media_upload(&mut self) -> Option<MediaUpload<dyn MediaUploadStream>>;
456    fn take_media_download(&mut self) -> Option<MediaDownload<dyn MediaDownloadStream>>;
457    fn get_media_download(&mut self) -> Option<&MediaDownload<dyn MediaDownloadStream>>;
458    fn has_resumable_media_upload(&self) -> bool;
459    fn take_resumable_media_upload(&mut self) -> Option<ResumableBody>;
460    fn get_request_method() -> Method;
461}
462pub trait ApiCallBuilder<'a, H: Hub> {
463    fn new(hub: &'a H) -> Self;
464}
465
466pub trait ConnectRequirements: Connect + Clone + Send + Sync + Debug + 'static {}
467impl<T> ConnectRequirements for T where T: Connect + Clone + Send + Sync + Debug + 'static {}
468
469pub trait Hub: HubBase + Debug + Clone {
470    fn call<'a, 'b, T>(
471        &'a self,
472        api_call: impl ApiCall<'b, T, Self> + 'b,
473    ) -> impl Future<Output = Result<T>>
474    where
475        T: Debug + DeserializeOwned,
476        Self: Sized,
477        'a: 'b,
478    {
479        call_with_hub(api_call, self)
480    }
481    fn get_base_url(&self) -> &str {
482        match self.get_region() {
483            None => self.get_global_url(),
484            Some(region) => self.get_url_for_region(region),
485        }
486    }
487    fn get_url_for_region(&self, region: &str) -> &str {
488        let endpoints = Self::get_endpoints();
489        match endpoints.get(region) {
490            None => self.get_global_url(),
491            Some(url) => url,
492        }
493    }
494}
495async fn call_with_hub<'a, T: Debug + DeserializeOwned, H: Hub + 'a>(
496    mut call: impl ApiCall<'a, T, H>,
497    hub: &'a H,
498) -> Result<T> {
499    let x = {
500        call.set_hub(hub);
501        call.call().await?
502    };
503    Ok(x)
504}
505impl<T: HubBase + Debug + Clone> Hub for T {}
506pub trait HubBase {
507    type Connector: ConnectRequirements;
508    type ConnectorAuth: ConnectRequirements;
509    fn new(
510        client: Client<Self::Connector, OutgoingBodyContent>,
511        auth: Authenticator<Self::ConnectorAuth>,
512    ) -> Self
513    where
514        Self: Sized;
515    fn get_client(&self) -> &Client<Self::Connector, OutgoingBodyContent>;
516    fn get_global_url(&self) -> &str;
517    fn get_service_path(&self) -> &str;
518    fn get_endpoints() -> &'static phf::Map<&'static str, &'static str>;
519    fn get_token(&self, scopes: &[impl Scope]) -> impl Future<Output = Result<String>>;
520    fn set_region(&mut self, region: impl Into<String>);
521    fn get_region(&self) -> Option<&String>;
522}
523
524pub trait Scope: AsRef<str> {}
525impl<T: AsRef<str>> Scope for T {}
gapirs_common/api.rs

gapirs_common/
api.rs
