feat(service/gdrive): add gdrive list support

core/src/services/gdrive/backend.rs

@@ -24,6 +24,7 @@ use http::StatusCode;
 
 use super::core::GdriveCore;
 use super::error::parse_error;
+use super::pager::GdrivePager;
 use super::writer::GdriveWriter;
 use crate::raw::*;
 use crate::services::gdrive::core::GdriveFile;
@@ -42,7 +43,7 @@ impl Accessor for GdriveBackend {
     type BlockingReader = ();
     type Writer = GdriveWriter;
     type BlockingWriter = ();
-    type Pager = ();
+    type Pager = GdrivePager;
     type BlockingPager = ();
 
     fn info(&self) -> AccessorInfo {
@@ -54,6 +55,10 @@ impl Accessor for GdriveBackend {
 
                 read: true,
 
+                list: true,
+
+                list_with_delimiter_slash: true,
+
                 write: true,
 
                 create_dir: true,
@@ -244,6 +249,13 @@ impl Accessor for GdriveBackend {
             Err(e)
         }
     }
+
+    async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Pager)> {
+        Ok((
+            RpList::default(),
+            GdrivePager::new(path.into(), self.core.clone()),
+        ))
+    }
 }
 
 impl GdriveBackend {

core/src/services/gdrive/core.rs

@@ -24,6 +24,7 @@ use bytes;
 use bytes::Bytes;
 use chrono::DateTime;
 use chrono::Utc;
+use futures::stream;
 use http::header;
 use http::Request;
 use http::Response;
@@ -99,7 +100,7 @@ impl GdriveCore {
                 "name = \"{}\" and \"{}\" in parents and trashed = false",
                 item, parent_id
             );
-            if i != file_path_items.len() - 1 {
+            if i != file_path_items.len() - 1 || path.ends_with('/') {
                 query += " and mimeType = 'application/vnd.google-apps.folder'";
             }
 
@@ -325,6 +326,67 @@ impl GdriveCore {
         self.client.send(req).await
     }
 
+    pub async fn gdrive_list(
+        &self,
+        path: &str,
+        page_size: i32,
+        next_page_token: Option<String>,
+    ) -> Result<Response<IncomingAsyncBody>> {
+        let file_id = self.get_file_id_by_path(path).await;
+
+        // when list over a no exist dir, `get_file_id_by_path` will return a NotFound Error, we should return a empty list in this case.
+        let q = match file_id {
+            Ok(file_id) => {
+                format!("'{}' in parents and trashed = false", file_id)
+            }
+            Err(e) => match e.kind() {
+                ErrorKind::NotFound => {
+                    return Response::builder()
+                        .status(StatusCode::OK)
+                        .body(IncomingAsyncBody::new(
+                            Box::new(oio::into_stream(stream::empty())),
+                            Some(0),
+                        ))
+                        .map_err(|e| {
+                            Error::new(
+                                ErrorKind::Unexpected,
+                                &format!("failed to create a empty response for list: {}", e),
+                            )
+                            .set_source(e)
+                        });
+                }
+                _ => {
+                    return Err(e);
+                }
+            },
+        };
+
+        let url = match next_page_token {
+            Some(page_token) => {
+                format!(
+                    "https://www.googleapis.com/drive/v3/files?pageSize={}&pageToken={}&q={}",
+                    page_size,
+                    page_token,
+                    percent_encode_path(q.as_str())
+                )
+            }
+            None => {
+                format!(
+                    "https://www.googleapis.com/drive/v3/files?pageSize={}&q={}",
+                    page_size,
+                    percent_encode_path(q.as_str())
+                )
+            }
+        };
+
+        let mut req = Request::get(&url)
+            .body(AsyncBody::Empty)
+            .map_err(new_request_build_error)?;
+        self.sign(&mut req).await?;
+
+        self.client.send(req).await
+    }
+
     // Update with content and metadata
     pub async fn gdrive_patch_metadata_request(
         &self,
@@ -556,4 +618,5 @@ pub struct GdriveFile {
 #[serde(rename_all = "camelCase")]
 pub(crate) struct GdriveFileList {
     pub(crate) files: Vec<GdriveFile>,
+    pub(crate) next_page_token: Option<String>,
 }

core/src/services/gdrive/error.rs

@@ -17,18 +17,29 @@
 
 use http::Response;
 use http::StatusCode;
+use serde::Deserialize;
 
 use crate::raw::*;
 use crate::Error;
 use crate::ErrorKind;
 use crate::Result;
 
+#[derive(Default, Debug, Deserialize)]
+struct GdriveError {
+    error: GdriveInnerError,
+}
+
+#[derive(Default, Debug, Deserialize)]
+struct GdriveInnerError {
+    message: String,
+}
+
 /// Parse error response into Error.
 pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
     let (parts, body) = resp.into_parts();
     let bs = body.bytes().await?;
 
-    let (kind, retryable) = match parts.status {
+    let (mut kind, mut retryable) = match parts.status {
         StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
         StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
         StatusCode::INTERNAL_SERVER_ERROR
@@ -38,7 +49,14 @@ pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
         _ => (ErrorKind::Unexpected, false),
     };
 
-    let message = String::from_utf8_lossy(&bs);
+    let (message, gdrive_err) = serde_json::from_slice::<GdriveError>(bs.as_ref())
+        .map(|gdrive_err| (format!("{gdrive_err:?}"), Some(gdrive_err)))
+        .unwrap_or_else(|_| (String::from_utf8_lossy(&bs).into_owned(), None));
+
+    if let Some(gdrive_err) = gdrive_err {
+        (kind, retryable) =
+            parse_gdrive_error_code(gdrive_err.error.message.as_str()).unwrap_or((kind, retryable));
+    }
 
     let mut err = Error::new(kind, &message);
 
@@ -50,3 +68,13 @@ pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
 
     Ok(err)
 }
+
+pub fn parse_gdrive_error_code(message: &str) -> Option<(ErrorKind, bool)> {
+    match message {
+        // > Please reduce your request rate.
+        //
+        // It's Ok to retry since later on the request rate may get reduced.
+        "User rate limit exceeded." => Some((ErrorKind::RateLimited, true)),
+        _ => None,
+    }
+}

core/src/services/gdrive/mod.rs

@@ -21,4 +21,5 @@ mod core;
 mod error;
 
 pub use builder::GdriveBuilder as Gdrive;
+mod pager;
 mod writer;

core/src/services/gdrive/pager.rs

@@ -0,0 +1,103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use crate::{
+    raw::{
+        build_rel_path, build_rooted_abs_path, new_json_deserialize_error,
+        oio::{self},
+    },
+    EntryMode, Metadata, Result,
+};
+use async_trait::async_trait;
+use http::StatusCode;
+
+use super::{
+    core::{GdriveCore, GdriveFileList},
+    error::parse_error,
+};
+pub struct GdrivePager {
+    path: String,
+    core: Arc<GdriveCore>,
+    next_page_token: Option<String>,
+    done: bool,
+}
+
+impl GdrivePager {
+    pub fn new(path: String, core: Arc<GdriveCore>) -> Self {
+        Self {
+            path,
+            core,
+            next_page_token: None,
+            done: false,
+        }
+    }
+}
+
+#[async_trait]
+impl oio::Page for GdrivePager {
+    async fn next(&mut self) -> Result<Option<Vec<oio::Entry>>> {
+        if self.done {
+            return Ok(None);
+        }
+
+        let resp = self
+            .core
+            .gdrive_list(&self.path, 100, self.next_page_token.clone())
+            .await?;
+
+        let bytes = match resp.status() {
+            StatusCode::OK => resp.into_body().bytes().await?,
+            _ => return Err(parse_error(resp).await?),
+        };
+
+        if bytes.is_empty() {
+            return Ok(None);
+        }
+
+        let decoded_response =
+            serde_json::from_slice::<GdriveFileList>(&bytes).map_err(new_json_deserialize_error)?;
+
+        if let Some(next_page_token) = decoded_response.next_page_token {
+            self.next_page_token = Some(next_page_token);
+        } else {
+            self.done = true;
+        }
+
+        let entries: Vec<oio::Entry> = decoded_response
+            .files
+            .into_iter()
+            .map(|mut file| {
+                let file_type = if file.mime_type.as_str() == "application/vnd.google-apps.folder" {
+                    file.name = format!("{}/", file.name);
+                    EntryMode::DIR
+                } else {
+                    EntryMode::FILE
+                };
+
+                let root = &self.core.root;
+                let path = format!("{}{}", build_rooted_abs_path(root, &self.path), file.name);
+                let normalized_path = build_rel_path(root, &path);
+
+                oio::Entry::new(&normalized_path, Metadata::new(file_type))
+            })
+            .collect();
+
+        Ok(Some(entries))
+    }
+}