feat(core): implement if_modified_since and if_unmodified_since for r…

…ead_with and reader_with (#5500)

* feat(core): implement if_modified_since and if_unmodified_since for read_with and reader_with

* fix time format

* fix cr comments

* fix tests

core/src/raw/chrono_util.rs

@@ -23,7 +23,7 @@ use chrono::Utc;
 
 use crate::*;
 
-/// Parse dateimt from rfc2822.
+/// Parse datetime from rfc2822.
 ///
 /// For example: `Fri, 28 Nov 2014 21:00:09 +0900`
 pub fn parse_datetime_from_rfc2822(s: &str) -> Result<DateTime<Utc>> {
@@ -34,7 +34,7 @@ pub fn parse_datetime_from_rfc2822(s: &str) -> Result<DateTime<Utc>> {
         })
 }
 
-/// Parse dateimt from rfc3339.
+/// Parse datetime from rfc3339.
 ///
 /// For example: `2014-11-28T21:00:09+09:00`
 pub fn parse_datetime_from_rfc3339(s: &str) -> Result<DateTime<Utc>> {
@@ -62,3 +62,24 @@ pub fn parse_datetime_from_from_timestamp(s: i64) -> Result<DateTime<Utc>> {
 
     Ok(st.into())
 }
+
+/// format datetime into http date, this format is required by:
+/// https://httpwg.org/specs/rfc9110.html#field.if-modified-since
+pub fn format_datetime_into_http_date(s: DateTime<Utc>) -> String {
+    s.format("%a, %d %b %Y %H:%M:%S GMT").to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_format_datetime_into_http_date() {
+        let s = "Sat, 29 Oct 1994 19:43:31 +0000";
+        let v = parse_datetime_from_rfc2822(s).unwrap();
+        assert_eq!(
+            format_datetime_into_http_date(v),
+            "Sat, 29 Oct 1994 19:43:31 GMT"
+        );
+    }
+}

core/src/raw/http_util/body.rs

@@ -39,7 +39,7 @@ pub struct HttpBody {
 
 /// # Safety
 ///
-/// HttpBody is send on non wasm32 targets.
+/// HttpBody is `Send` on non wasm32 targets.
 unsafe impl Send for HttpBody {}
 
 /// # Safety

core/src/raw/ops.rs

@@ -19,11 +19,11 @@
 //!
 //! By using ops, users can add more context for operation.
 
-use std::collections::HashMap;
-use std::time::Duration;
-
 use crate::raw::*;
 use crate::*;
+use chrono::{DateTime, Utc};
+use std::collections::HashMap;
+use std::time::Duration;
 
 /// Args for `create` operation.
 ///
@@ -299,6 +299,8 @@ pub struct OpRead {
     range: BytesRange,
     if_match: Option<String>,
     if_none_match: Option<String>,
+    if_modified_since: Option<DateTime<Utc>>,
+    if_unmodified_since: Option<DateTime<Utc>>,
     override_content_type: Option<String>,
     override_cache_control: Option<String>,
     override_content_disposition: Option<String>,
@@ -384,6 +386,28 @@ impl OpRead {
         self.if_none_match.as_deref()
     }
 
+    /// Set the If-Modified-Since of the option
+    pub fn with_if_modified_since(mut self, v: DateTime<Utc>) -> Self {
+        self.if_modified_since = Some(v);
+        self
+    }
+
+    /// Get If-Modified-Since from option
+    pub fn if_modified_since(&self) -> Option<DateTime<Utc>> {
+        self.if_modified_since
+    }
+
+    /// Set the If-Unmodified-Since of the option
+    pub fn with_if_unmodified_since(mut self, v: DateTime<Utc>) -> Self {
+        self.if_unmodified_since = Some(v);
+        self
+    }
+
+    /// Get If-Unmodified-Since from option
+    pub fn if_unmodified_since(&self) -> Option<DateTime<Utc>> {
+        self.if_unmodified_since
+    }
+
     /// Set the version of the option
     pub fn with_version(mut self, version: &str) -> Self {
         self.version = Some(version.to_string());

core/src/services/s3/backend.rs

@@ -944,6 +944,8 @@ impl Access for S3Backend {
                 read: true,
                 read_with_if_match: true,
                 read_with_if_none_match: true,
+                read_with_if_modified_since: true,
+                read_with_if_unmodified_since: true,
                 read_with_override_cache_control: true,
                 read_with_override_content_disposition: true,
                 read_with_override_content_type: true,

core/src/services/s3/core.rs

@@ -28,7 +28,6 @@ use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::Bytes;
 use constants::X_AMZ_META_PREFIX;
-use http::header::HeaderName;
 use http::header::CACHE_CONTROL;
 use http::header::CONTENT_DISPOSITION;
 use http::header::CONTENT_ENCODING;
@@ -37,6 +36,7 @@ use http::header::CONTENT_TYPE;
 use http::header::HOST;
 use http::header::IF_MATCH;
 use http::header::IF_NONE_MATCH;
+use http::header::{HeaderName, IF_MODIFIED_SINCE, IF_UNMODIFIED_SINCE};
 use http::HeaderValue;
 use http::Request;
 use http::Response;
@@ -406,6 +406,21 @@ impl S3Core {
         if let Some(if_match) = args.if_match() {
             req = req.header(IF_MATCH, if_match);
         }
+
+        if let Some(if_modified_since) = args.if_modified_since() {
+            req = req.header(
+                IF_MODIFIED_SINCE,
+                format_datetime_into_http_date(if_modified_since),
+            );
+        }
+
+        if let Some(if_unmodified_since) = args.if_unmodified_since() {
+            req = req.header(
+                IF_UNMODIFIED_SINCE,
+                format_datetime_into_http_date(if_unmodified_since),
+            );
+        }
+
         // Set SSE headers.
         // TODO: how will this work with presign?
         req = self.insert_sse_headers(req, false);

core/src/types/capability.rs

@@ -107,6 +107,10 @@ pub struct Capability {
     pub read_with_if_match: bool,
     /// Indicates if conditional read operations using If-None-Match are supported.
     pub read_with_if_none_match: bool,
+    /// Indicates if conditional read operations using If-Modified-Since are supported.
+    pub read_with_if_modified_since: bool,
+    /// Indicates if conditional read operations using If-Unmodified-Since are supported.
+    pub read_with_if_unmodified_since: bool,
     /// Indicates if Cache-Control header override is supported during read operations.
     pub read_with_override_cache_control: bool,
     /// Indicates if Content-Disposition header override is supported during read operations.

core/src/types/operator/operator_futures.rs

@@ -19,13 +19,13 @@
 //!
 //! By using futures, users can add more options for operation.
 
+use chrono::{DateTime, Utc};
+use futures::Future;
 use std::collections::HashMap;
 use std::future::IntoFuture;
 use std::ops::RangeBounds;
 use std::time::Duration;
 
-use futures::Future;
-
 use crate::raw::*;
 use crate::*;
 
@@ -212,6 +212,16 @@ impl<F: Future<Output = Result<Buffer>>> FutureRead<F> {
         self.map(|(args, op_reader)| (args.with_if_none_match(v), op_reader))
     }
 
+    /// Set the If-Modified-Since for this operation.
+    pub fn if_modified_since(self, v: DateTime<Utc>) -> Self {
+        self.map(|(args, op_reader)| (args.with_if_modified_since(v), op_reader))
+    }
+
+    /// Set the If-Unmodified-Since for this operation.
+    pub fn if_unmodified_since(self, v: DateTime<Utc>) -> Self {
+        self.map(|(args, op_reader)| (args.with_if_unmodified_since(v), op_reader))
+    }
+
     /// Set the version for this operation.
     pub fn version(self, v: &str) -> Self {
         self.map(|(args, op_reader)| (args.with_version(v), op_reader))
@@ -258,6 +268,16 @@ impl<F: Future<Output = Result<Reader>>> FutureReader<F> {
         self.map(|(op_read, op_reader)| (op_read.with_if_none_match(etag), op_reader))
     }
 
+    /// Set the If-Modified-Since for this operation.
+    pub fn if_modified_since(self, v: DateTime<Utc>) -> Self {
+        self.map(|(op_read, op_reader)| (op_read.with_if_modified_since(v), op_reader))
+    }
+
+    /// Set the If-Unmodified-Since for this operation.
+    pub fn if_unmodified_since(self, v: DateTime<Utc>) -> Self {
+        self.map(|(op_read, op_reader)| (op_read.with_if_unmodified_since(v), op_reader))
+    }
+
     /// Set the version for this operation.
     pub fn version(self, v: &str) -> Self {
         self.map(|(op_read, op_reader)| (op_read.with_version(v), op_reader))

core/tests/behavior/async_read.rs

@@ -18,15 +18,15 @@
 use std::str::FromStr;
 use std::time::Duration;
 
+use crate::*;
 use futures::AsyncReadExt;
 use futures::TryStreamExt;
 use http::StatusCode;
 use log::warn;
 use reqwest::Url;
 use sha2::Digest;
 use sha2::Sha256;
-
-use crate::*;
+use tokio::time::sleep;
 
 pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
     let cap = op.info().full_capability();
@@ -39,9 +39,13 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_reader,
             test_reader_with_if_match,
             test_reader_with_if_none_match,
+            test_reader_with_if_modified_since,
+            test_reader_with_if_unmodified_since,
             test_read_not_exist,
             test_read_with_if_match,
             test_read_with_if_none_match,
+            test_read_with_if_modified_since,
+            test_read_with_if_unmodified_since,
             test_read_with_dir_path,
             test_read_with_special_chars,
             test_read_with_override_cache_control,
@@ -270,6 +274,64 @@ pub async fn test_reader_with_if_none_match(op: Operator) -> anyhow::Result<()>
     Ok(())
 }
 
+/// Reader with if_modified_since should match, otherwise, a ConditionNotMatch error will be returned.
+pub async fn test_reader_with_if_modified_since(op: Operator) -> anyhow::Result<()> {
+    if !op.info().full_capability().read_with_if_modified_since {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+    let last_modified_time = op.stat(&path).await?.last_modified().unwrap();
+
+    let since = last_modified_time - chrono::Duration::seconds(1);
+    let reader = op.reader_with(&path).if_modified_since(since).await?;
+    let bs = reader.read(..).await?.to_bytes();
+    assert_eq!(bs, content);
+
+    sleep(Duration::from_secs(1)).await;
+
+    let since = last_modified_time + chrono::Duration::seconds(1);
+    let reader = op.reader_with(&path).if_modified_since(since).await?;
+    let res = reader.read(..).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
+
+/// Reader with if_unmodified_since should match, otherwise, a ConditionNotMatch error will be returned.
+pub async fn test_reader_with_if_unmodified_since(op: Operator) -> anyhow::Result<()> {
+    if !op.info().full_capability().read_with_if_unmodified_since {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+    let last_modified_time = op.stat(&path).await?.last_modified().unwrap();
+
+    let since = last_modified_time - chrono::Duration::seconds(1);
+    let reader = op.reader_with(&path).if_unmodified_since(since).await?;
+    let res = reader.read(..).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    sleep(Duration::from_secs(1)).await;
+
+    let since = last_modified_time + chrono::Duration::seconds(1);
+    let reader = op.reader_with(&path).if_unmodified_since(since).await?;
+    let bs = reader.read(..).await?.to_bytes();
+    assert_eq!(bs, content);
+
+    Ok(())
+}
+
 /// Read with if_none_match should match, else get a ConditionNotMatch error.
 pub async fn test_read_with_if_none_match(op: Operator) -> anyhow::Result<()> {
     if !op.info().full_capability().read_with_if_none_match {
@@ -492,6 +554,68 @@ pub async fn test_read_with_override_content_type(op: Operator) -> anyhow::Resul
     Ok(())
 }
 
+/// Read with if_modified_since should match, otherwise, a ConditionNotMatch error will be returned.
+pub async fn test_read_with_if_modified_since(op: Operator) -> anyhow::Result<()> {
+    if !op.info().full_capability().read_with_if_modified_since {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+    let last_modified_time = op.stat(&path).await?.last_modified().unwrap();
+
+    let since = last_modified_time - chrono::Duration::seconds(1);
+    let bs = op
+        .read_with(&path)
+        .if_modified_since(since)
+        .await?
+        .to_bytes();
+    assert_eq!(bs, content);
+
+    sleep(Duration::from_secs(1)).await;
+
+    let since = last_modified_time + chrono::Duration::seconds(1);
+    let res = op.read_with(&path).if_modified_since(since).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
+
+/// Read with if_unmodified_since should match, otherwise, a ConditionNotMatch error will be returned.
+pub async fn test_read_with_if_unmodified_since(op: Operator) -> anyhow::Result<()> {
+    if !op.info().full_capability().read_with_if_unmodified_since {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+    let last_modified = op.stat(&path).await?.last_modified().unwrap();
+
+    let since = last_modified - chrono::Duration::seconds(3600);
+    let res = op.read_with(&path).if_unmodified_since(since).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    sleep(Duration::from_secs(1)).await;
+
+    let since = last_modified + chrono::Duration::seconds(1);
+    let bs = op
+        .read_with(&path)
+        .if_unmodified_since(since)
+        .await?
+        .to_bytes();
+    assert_eq!(bs, content);
+
+    Ok(())
+}
+
 /// Read full content should match.
 pub async fn test_read_only_read_full(op: Operator) -> anyhow::Result<()> {
     let bs = op.read("normal_file.txt").await?.to_bytes();