rs_transfer 8.0.0

use crate::{
  endpoint::GcsEndpoint,
  error::Error,
  list::{TreeItem, TreeItemKind, TreeItems, TreeList, join_path, strip_relative_path},
};
use cloud_storage::{ListRequest, Object};
use futures::StreamExt;
use std::{convert::TryFrom, time::SystemTime};

const PATH_SEPARATOR: char = '/';

#[async_trait::async_trait]
impl TreeList for GcsEndpoint {
  async fn list(&self, root_path: &str, prefix: Option<&str>) -> Result<TreeItems, Error> {
    let delimiter = Self::get_delimiter(PATH_SEPARATOR);
    let request_prefix = Self::get_request_prefix(root_path, prefix);
    let root_path = Self::get_root_path(root_path, PATH_SEPARATOR);

    let request = ListRequest {
      delimiter: Some(delimiter.clone()),
      prefix: request_prefix,
      ..Default::default()
    };

    let client = self.connection();
    let root_path = root_path.as_str();

    let items = match client.object().list(self.bucket(), request).await {
      Ok(list_stream) => {
        let object_lists = list_stream
          .filter_map(|object_list| async move {
            object_list
              .map(|list| {
                list
                  .items
                  .into_iter()
                  .filter_map(|object| {
                    let item = TreeItem::try_from((root_path, &object));

                    if let Err(error) = &item {
                      log::warn!("Cannot access entry in {}: {:?}", self.bucket(), error);
                    }

                    item.ok()
                  })
                  .chain(
                    list
                      .prefixes
                      .into_iter()
                      .map(|prefix| GcsPrefix { prefix })
                      .filter_map(|prefix| {
                        let item = TreeItem::try_from((root_path, &prefix));

                        if let Err(error) = &item {
                          log::warn!("Cannot access entry in {}: {:?}", self.bucket(), error);
                        }

                        item.ok()
                      }),
                  )
                  .collect::<TreeItems>()
              })
              .ok()
          })
          .collect::<Vec<TreeItems>>() // this first collect is due to following bug in Rust: https://github.com/rust-lang/rust/issues/102211
          .await;

        object_lists.into_iter().flatten().collect::<TreeItems>()
      }
      Err(error) => {
        log::warn!("Cannot access {} content: {:?}", self.bucket(), error);
        TreeItems::default()
      }
    };

    Ok(items)
  }
}

impl GcsEndpoint {
  fn get_request_prefix(root_path: &str, prefix: Option<&str>) -> Option<String> {
    if (root_path.is_empty() || root_path == PATH_SEPARATOR.to_string()) && prefix.is_none() {
      None
    } else {
      Some(format!(
        "{}{}",
        Self::get_absolute_prefix(root_path, prefix, PATH_SEPARATOR)
          .trim_start_matches(PATH_SEPARATOR),
        PATH_SEPARATOR,
      ))
    }
  }
}

struct GcsPrefix {
  prefix: String,
}

impl TryFrom<(&str, &Object)> for TreeItem {
  type Error = Error;

  fn try_from((root_path, object): (&str, &Object)) -> Result<Self, Self::Error> {
    let path = strip_relative_path(root_path, &object.name, PATH_SEPARATOR);

    Ok(Self::new(
      TreeItemKind::File,
      &path,
      root_path,
      object.size,
      SystemTime::from(object.updated),
      object.content_type.clone(),
      PATH_SEPARATOR,
    ))
  }
}

impl TryFrom<(&str, &GcsPrefix)> for TreeItem {
  type Error = Error;

  fn try_from((root_path, prefix): (&str, &GcsPrefix)) -> Result<Self, Self::Error> {
    let path = join_path(
      root_path,
      prefix.prefix.trim_end_matches(PATH_SEPARATOR),
      PATH_SEPARATOR,
    );

    Ok(Self::new(
      TreeItemKind::Folder,
      &path,
      root_path,
      0,
      SystemTime::UNIX_EPOCH,
      None,
      '/',
    ))
  }
}

#[test]
pub fn test_get_root_path() {
  assert_eq!("/", &GcsEndpoint::get_root_path("", PATH_SEPARATOR));
  assert_eq!("/", &GcsEndpoint::get_root_path("/", PATH_SEPARATOR));
  assert_eq!("/root", &GcsEndpoint::get_root_path("root", PATH_SEPARATOR));
  assert_eq!(
    "/root",
    &GcsEndpoint::get_root_path("root/", PATH_SEPARATOR)
  );
  assert_eq!(
    "/root",
    &GcsEndpoint::get_root_path("/root", PATH_SEPARATOR)
  );
  assert_eq!(
    "/root",
    &GcsEndpoint::get_root_path("/root/", PATH_SEPARATOR)
  );
  assert_eq!(
    "/root/path",
    &GcsEndpoint::get_root_path("root/path", PATH_SEPARATOR)
  );
  assert_eq!(
    "/root/path",
    &GcsEndpoint::get_root_path("root/path/", PATH_SEPARATOR)
  );
  assert_eq!(
    "/root/path",
    &GcsEndpoint::get_root_path("/root/path", PATH_SEPARATOR)
  );
  assert_eq!(
    "/root/path",
    &GcsEndpoint::get_root_path("/root/path/", PATH_SEPARATOR)
  );
}

#[test]
pub fn test_list_request_prefix() {
  assert_eq!(None, GcsEndpoint::get_request_prefix("", None));
  assert_eq!(None, GcsEndpoint::get_request_prefix("/", None));
  assert_eq!(
    Some("root/".to_string()),
    GcsEndpoint::get_request_prefix("/root", None)
  );
  assert_eq!(
    Some("root/path/".to_string()),
    GcsEndpoint::get_request_prefix("/root/path", None)
  );

  assert_eq!(
    Some("prefix/".to_string()),
    GcsEndpoint::get_request_prefix("", Some("prefix"))
  );
  assert_eq!(
    Some("prefix/".to_string()),
    GcsEndpoint::get_request_prefix("/", Some("prefix"))
  );
  assert_eq!(
    Some("root/prefix/".to_string()),
    GcsEndpoint::get_request_prefix("/root", Some("prefix"))
  );
  assert_eq!(
    Some("root/path/prefix/".to_string()),
    GcsEndpoint::get_request_prefix("/root/path", Some("prefix"))
  );
}