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
48// OneDrive requires multipart writes to specify the total size of the file.
49// While OpenDAL supports multiple writes, due to OneDrive's limitation, we opt to
50// use `OneShotWrite` instead of `MultipartWrite`.
51impl oio::OneShotWrite for OneDriveWriter {
52    async fn write_once(&self, bs: Buffer) -> Result<Metadata> {
53        let size = bs.len();
54
55        let meta = if size <= Self::MAX_SIMPLE_SIZE {
56            self.write_simple(bs).await?
57        } else {
58            self.write_chunked(bs).await?
59        };
60
61        Ok(meta)
62    }
63}
64
65impl OneDriveWriter {
66    async fn write_simple(&self, bs: Buffer) -> Result<Metadata> {
67        let response = self
68            .core
69            .onedrive_upload_simple(&self.path, &self.op, bs)
70            .await?;
71
72        match response.status() {
73            StatusCode::CREATED | StatusCode::OK => {
74                let item: OneDriveItem = serde_json::from_reader(response.into_body().reader())
75                    .map_err(new_json_deserialize_error)?;
76
77                let mut meta = Metadata::new(EntryMode::FILE)
78                    .with_etag(item.e_tag)
79                    .with_content_length(item.size.max(0) as u64);
80
81                let last_modified = item.last_modified_date_time;
82                let date_utc_last_modified = parse_datetime_from_rfc3339(&last_modified)?;
83                meta.set_last_modified(date_utc_last_modified);
84
85                Ok(meta)
86            }
87            _ => Err(parse_error(response)),
88        }
89    }
90
91    pub(crate) async fn write_chunked(&self, bs: Buffer) -> Result<Metadata> {
92        // 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
93        // 1. Create an upload session
94        // 2. Upload the bytes of each chunk
95        // 3. Commit the session
96
97        let session_response = self.create_upload_session().await?;
98
99        let mut offset = 0;
100        let total_bytes = bs.to_bytes();
101        let total_len = total_bytes.len();
102        let chunks = total_bytes.chunks(OneDriveWriter::CHUNK_SIZE_FACTOR);
103
104        for chunk in chunks {
105            let mut end = offset + OneDriveWriter::CHUNK_SIZE_FACTOR;
106            if end > total_bytes.len() {
107                end = total_bytes.len();
108            }
109            let chunk_end = end - 1;
110
111            let response = self
112                .core
113                .onedrive_chunked_upload(
114                    &session_response.upload_url,
115                    &self.op,
116                    offset,
117                    chunk_end,
118                    total_len,
119                    Buffer::from(Bytes::copy_from_slice(chunk)),
120                )
121                .await?;
122
123            match response.status() {
124                // Typical response code: 202 Accepted
125                // Reference: https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_put_content?view=odsp-graph-online#response
126                StatusCode::ACCEPTED | StatusCode::OK => {} // skip, in the middle of upload
127                StatusCode::CREATED => {
128                    // last trunk
129                    let item: OneDriveItem = serde_json::from_reader(response.into_body().reader())
130                        .map_err(new_json_deserialize_error)?;
131
132                    let mut meta = Metadata::new(EntryMode::FILE)
133                        .with_etag(item.e_tag)
134                        .with_content_length(item.size.max(0) as u64);
135
136                    let last_modified = item.last_modified_date_time;
137                    let date_utc_last_modified = parse_datetime_from_rfc3339(&last_modified)?;
138                    meta.set_last_modified(date_utc_last_modified);
139                    return Ok(meta);
140                }
141                _ => return Err(parse_error(response)),
142            }
143
144            offset += OneDriveWriter::CHUNK_SIZE_FACTOR;
145        }
146
147        debug_assert!(false, "should have returned");
148
149        Ok(Metadata::default()) // should not happen, but start with handling this gracefully - do nothing, but return the default metadata
150    }
151
152    async fn create_upload_session(&self) -> Result<OneDriveUploadSessionCreationResponseBody> {
153        let response = self
154            .core
155            .onedrive_create_upload_session(&self.path, &self.op)
156            .await?;
157        match response.status() {
158            StatusCode::OK => {
159                let bs = response.into_body();
160                let result: OneDriveUploadSessionCreationResponseBody =
161                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
162                Ok(result)
163            }
164            _ => Err(parse_error(response)),
165        }
166    }
167}
opendal/services/onedrive/writer.rs

opendal/services/onedrive/
writer.rs
