Add HttpStore (#3294)

.github/workflows/object_store.yml

@@ -77,6 +77,7 @@ jobs:
       AZURE_USE_EMULATOR: "1"
       AZURITE_BLOB_STORAGE_URL: "http://localhost:10000"
       AZURITE_QUEUE_STORAGE_URL: "http://localhost:10001"
+      HTTP_URL: "http://localhost:8080"
       GOOGLE_SERVICE_ACCOUNT: "/tmp/gcs.json"
       OBJECT_STORE_BUCKET: test-bucket
 
@@ -91,6 +92,9 @@ jobs:
           curl -v -X POST --data-binary '{"name":"test-bucket"}' -H "Content-Type: application/json" "http://localhost:4443/storage/v1/b"
           echo '{"gcs_base_url": "http://localhost:4443", "disable_oauth": true, "client_email": "", "private_key": ""}' > "$GOOGLE_SERVICE_ACCOUNT"
 
+      - name: Setup WebDav
+        run: docker run -d -p 8080:80 rclone/rclone serve webdav /data --addr :80
+
       - name: Setup LocalStack (AWS emulation)
         env:
           AWS_DEFAULT_REGION: "us-east-1"
@@ -120,7 +124,7 @@ jobs:
           OBJECT_STORE_AWS_ACCESS_KEY_ID: test
           OBJECT_STORE_AWS_SECRET_ACCESS_KEY: test
           OBJECT_STORE_AWS_ENDPOINT: http://localhost:4566
-        run: cargo test -p object_store --features=aws,azure,gcp
+        run: cargo test -p object_store --features=aws,azure,gcp,http
 
   # test the object_store crate builds against wasm32 in stable rust
   wasm32-build:

object_store/Cargo.toml

@@ -61,6 +61,7 @@ cloud = ["serde", "serde_json", "quick-xml", "reqwest", "reqwest/json", "reqwest
 azure = ["cloud"]
 gcp = ["cloud", "rustls-pemfile"]
 aws = ["cloud"]
+http = ["cloud"]
 
 # Experimental support for AWS_PROFILE
 aws_profile = ["aws", "aws-config", "aws-types"]

object_store/src/azure/client.rs

@@ -20,20 +20,20 @@ use crate::azure::credential::*;
 use crate::client::pagination::stream_paginated;
 use crate::client::retry::RetryExt;
 use crate::path::DELIMITER;
-use crate::util::{format_http_range, format_prefix};
+use crate::util::{deserialize_rfc1123, format_http_range, format_prefix};
 use crate::{
     BoxStream, ClientOptions, ListResult, ObjectMeta, Path, Result, RetryConfig,
     StreamExt,
 };
 use bytes::{Buf, Bytes};
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::{DateTime, Utc};
 use itertools::Itertools;
 use reqwest::header::CONTENT_TYPE;
 use reqwest::{
     header::{HeaderValue, CONTENT_LENGTH, IF_NONE_MATCH, RANGE},
     Client as ReqwestClient, Method, Response, StatusCode,
 };
-use serde::{Deserialize, Deserializer, Serialize};
+use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
 use std::collections::HashMap;
 use std::ops::Range;
@@ -479,7 +479,7 @@ impl TryFrom<Blob> for ObjectMeta {
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
 #[serde(rename_all = "PascalCase")]
 struct BlobProperties {
-    #[serde(deserialize_with = "deserialize_http_date", rename = "Last-Modified")]
+    #[serde(deserialize_with = "deserialize_rfc1123", rename = "Last-Modified")]
     pub last_modified: DateTime<Utc>,
     pub etag: String,
     #[serde(rename = "Content-Length")]
@@ -492,16 +492,6 @@ struct BlobProperties {
     pub content_language: Option<String>,
 }
 
-// deserialize dates used in Azure payloads according to rfc1123
-fn deserialize_http_date<'de, D>(deserializer: D) -> Result<DateTime<Utc>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    let s = String::deserialize(deserializer)?;
-    Utc.datetime_from_str(&s, RFC1123_FMT)
-        .map_err(serde::de::Error::custom)
-}
-
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct BlockId(Bytes);
 

object_store/src/azure/mod.rs

@@ -46,6 +46,7 @@ use std::sync::Arc;
 use tokio::io::AsyncWrite;
 use url::Url;
 
+use crate::util::RFC1123_FMT;
 pub use credential::authority_hosts;
 
 mod client;
@@ -219,7 +220,7 @@ impl ObjectStore for MicrosoftAzure {
             .to_str()
             .context(BadHeaderSnafu)?;
         let last_modified = Utc
-            .datetime_from_str(last_modified, credential::RFC1123_FMT)
+            .datetime_from_str(last_modified, RFC1123_FMT)
             .context(InvalidLastModifiedSnafu { last_modified })?;
 
         let content_length = headers

object_store/src/client/mod.rs

@@ -20,8 +20,10 @@
 pub mod backoff;
 #[cfg(test)]
 pub mod mock_server;
+#[cfg(any(feature = "aws", feature = "gcp", feature = "azure"))]
 pub mod pagination;
 pub mod retry;
+#[cfg(any(feature = "aws", feature = "gcp", feature = "azure"))]
 pub mod token;
 
 use reqwest::header::{HeaderMap, HeaderValue};