opendal/services/onedrive/
writer.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::sync::Arc;
19
20use bytes::Buf;
21use bytes::Bytes;
22use http::StatusCode;
23
24use super::core::OneDriveCore;
25use super::error::parse_error;
26use super::graph_model::{OneDriveItem, OneDriveUploadSessionCreationResponseBody};
27use crate::raw::*;
28use crate::*;
29
30pub struct OneDriveWriter {
31    core: Arc<OneDriveCore>,
32    op: OpWrite,
33    path: String,
34}
35
36impl OneDriveWriter {
37    const MAX_SIMPLE_SIZE: usize = 4 * 1024 * 1024; // 4MB
38
39    // OneDrive demands the chunk size to be a multiple to to 320 KiB.
40    // Choose a value smaller than `MAX_SIMPLE_SIZE`
41    const CHUNK_SIZE_FACTOR: usize = 327_680 * 12; // floor(MAX_SIMPLE_SIZE / 320KB)
42
43    pub fn new(core: Arc<OneDriveCore>, op: OpWrite, path: String) -> Self {
44        OneDriveWriter { core, op, path }
45    }
46}
47
48impl oio::OneShotWrite for OneDriveWriter {
49    async fn write_once(&self, bs: Buffer) -> Result<Metadata> {
50        let size = bs.len();
51
52        let meta = if size <= Self::MAX_SIMPLE_SIZE {
53            self.write_simple(bs).await?
54        } else {
55            self.write_chunked(bs).await?
56        };
57
58        Ok(meta)
59    }
60}
61
62impl OneDriveWriter {
63    async fn write_simple(&self, bs: Buffer) -> Result<Metadata> {
64        let response = self
65            .core
66            .onedrive_upload_simple(&self.path, &self.op, bs)
67            .await?;
68
69        match response.status() {
70            StatusCode::CREATED | StatusCode::OK => {
71                let item: OneDriveItem = serde_json::from_reader(response.into_body().reader())
72                    .map_err(new_json_deserialize_error)?;
73
74                let mut meta = Metadata::new(EntryMode::FILE)
75                    .with_etag(item.e_tag)
76                    .with_content_length(item.size.max(0) as u64);
77
78                let last_modified = item.last_modified_date_time;
79                let date_utc_last_modified = parse_datetime_from_rfc3339(&last_modified)?;
80                meta.set_last_modified(date_utc_last_modified);
81
82                Ok(meta)
83            }
84            _ => Err(parse_error(response)),
85        }
86    }
87
88    pub(crate) async fn write_chunked(&self, bs: Buffer) -> Result<Metadata> {
89        // Upload large files via sessions: https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_createuploadsession?view=odsp-graph-online#upload-bytes-to-the-upload-session
90        // 1. Create an upload session
91        // 2. Upload the bytes of each chunk
92        // 3. Commit the session
93
94        let session_response = self.create_upload_session().await?;
95
96        let mut offset = 0;
97        let total_bytes = bs.to_bytes();
98        let total_len = total_bytes.len();
99        let chunks = total_bytes.chunks(OneDriveWriter::CHUNK_SIZE_FACTOR);
100
101        for chunk in chunks {
102            let mut end = offset + OneDriveWriter::CHUNK_SIZE_FACTOR;
103            if end > total_bytes.len() {
104                end = total_bytes.len();
105            }
106            let chunk_end = end - 1;
107
108            let response = self
109                .core
110                .onedrive_chunked_upload(
111                    &session_response.upload_url,
112                    &self.op,
113                    offset,
114                    chunk_end,
115                    total_len,
116                    Buffer::from(Bytes::copy_from_slice(chunk)),
117                )
118                .await?;
119
120            match response.status() {
121                // Typical response code: 202 Accepted
122                // Reference: https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_put_content?view=odsp-graph-online#response
123                StatusCode::ACCEPTED | StatusCode::OK => {} // skip, in the middle of upload
124                StatusCode::CREATED => {
125                    // last trunk
126                    let item: OneDriveItem = serde_json::from_reader(response.into_body().reader())
127                        .map_err(new_json_deserialize_error)?;
128
129                    let mut meta = Metadata::new(EntryMode::FILE)
130                        .with_etag(item.e_tag)
131                        .with_content_length(item.size.max(0) as u64);
132
133                    let last_modified = item.last_modified_date_time;
134                    let date_utc_last_modified = parse_datetime_from_rfc3339(&last_modified)?;
135                    meta.set_last_modified(date_utc_last_modified);
136                    return Ok(meta);
137                }
138                _ => return Err(parse_error(response)),
139            }
140
141            offset += OneDriveWriter::CHUNK_SIZE_FACTOR;
142        }
143
144        debug_assert!(false, "should have returned");
145
146        Ok(Metadata::default()) // should not happen, but start with handling this gracefully - do nothing, but return the default metadata
147    }
148
149    async fn create_upload_session(&self) -> Result<OneDriveUploadSessionCreationResponseBody> {
150        let response = self
151            .core
152            .onedrive_create_upload_session(&self.path, &self.op)
153            .await?;
154        match response.status() {
155            StatusCode::OK => {
156                let bs = response.into_body();
157                let result: OneDriveUploadSessionCreationResponseBody =
158                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
159                Ok(result)
160            }
161            _ => Err(parse_error(response)),
162        }
163    }
164}
opendal/services/onedrive/writer.rs

opendal/services/onedrive/
writer.rs
