Migrate API to async

* Uses Tokio 1.x and a recent reqwest
* Migrate a few examples
* Move some APIs from iterator to async stream

Co-authored-by: David Gräff <davgraeff@gmail.com>
Signed-off-by: Michał Kawalec <michal@monad.cat>
Signed-off-by: David Gräff <davgraeff@gmail.com>

Author: mkawalec

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "firestore-db-and-auth"
-version = "0.6.1"
+version = "0.8.0"
 authors = ["David Gräff <david.graeff@web.de>"]
 edition = "2018"
 license = "MIT"
@@ -12,13 +12,20 @@ maintenance = { status = "passively-maintained" }
 repository = "https://github.com/davidgraeff/firestore-db-and-auth-rs"
 
 [dependencies]
+bytes = "1.1"
+cache_control = "0.2"
 reqwest = { version = "0.11", default-features = false, features = ["json", "blocking"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 chrono = { version = "0.4", features = ["serde"] }
 biscuit = "0.5"
 ring = "0.16"
 base64 = "0.13"
+async-trait = "0.1"
+tokio = { version = "1.13", features = ["macros"] }
+futures = "0.3"
+pin-project = "1.0"
+http = "0.2"
 
 [dependencies.rocket]
 version = "0.4.6"

examples/create_read_write_document.rs

@@ -3,6 +3,8 @@ use firestore_db_and_auth::{documents, dto, errors, sessions, Credentials, Fireb
 use firestore_db_and_auth::documents::WriteResult;
 use serde::{Deserialize, Serialize};
 
+use futures::stream::StreamExt;
+
 mod utils;
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -21,7 +23,7 @@ struct DemoDTOPartial {
     an_int: u32,
 }
 
-fn write_document(session: &mut ServiceSession, doc_id: &str) -> errors::Result<WriteResult> {
+async fn write_document(session: &mut ServiceSession, doc_id: &str) -> errors::Result<WriteResult> {
     println!("Write document");
 
     let obj = DemoDTO {
@@ -30,10 +32,10 @@ fn write_document(session: &mut ServiceSession, doc_id: &str) -> errors::Result<
         a_timestamp: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Nanos, true),
     };
 
-    documents::write(session, "tests", Some(doc_id), &obj, documents::WriteOptions::default())
+    documents::write(session, "tests", Some(doc_id), &obj, documents::WriteOptions::default()).await
 }
 
-fn write_partial_document(session: &mut ServiceSession, doc_id: &str) -> errors::Result<WriteResult> {
+async fn write_partial_document(session: &mut ServiceSession, doc_id: &str) -> errors::Result<WriteResult> {
     println!("Partial write document");
 
     let obj = DemoDTOPartial {
@@ -48,6 +50,7 @@ fn write_partial_document(session: &mut ServiceSession, doc_id: &str) -> errors:
         &obj,
         documents::WriteOptions { merge: true },
     )
+    .await
 }
 
 fn check_write(result: WriteResult, doc_id: &str) {
@@ -62,25 +65,25 @@ fn check_write(result: WriteResult, doc_id: &str) {
     );
 }
 
-fn service_account_session(cred: Credentials) -> errors::Result<()> {
-    let mut session = ServiceSession::new(cred).unwrap();
-    let b = session.access_token().to_owned();
+async fn service_account_session(cred: Credentials) -> errors::Result<()> {
+    let mut session = ServiceSession::new(cred).await.unwrap();
+    let b = session.access_token().await.to_owned();
 
     let doc_id = "service_test";
-    check_write(write_document(&mut session, doc_id)?, doc_id);
+    check_write(write_document(&mut session, doc_id).await?, doc_id);
 
     // Check if cached value is used
-    assert_eq!(session.access_token(), b);
+    assert_eq!(session.access_token().await, b);
 
     println!("Read and compare document");
-    let read: DemoDTO = documents::read(&mut session, "tests", doc_id)?;
+    let read: DemoDTO = documents::read(&mut session, "tests", doc_id).await?;
 
     assert_eq!(read.a_string, "abcd");
     assert_eq!(read.an_int, 14);
 
-    check_write(write_partial_document(&mut session, doc_id)?, doc_id);
+    check_write(write_partial_document(&mut session, doc_id).await?, doc_id);
     println!("Read and compare document");
-    let read: DemoDTOPartial = documents::read(&mut session, "tests", doc_id)?;
+    let read: DemoDTOPartial = documents::read(&mut session, "tests", doc_id).await?;
 
     // Should be updated
     assert_eq!(read.an_int, 16);
@@ -90,14 +93,15 @@ fn service_account_session(cred: Credentials) -> errors::Result<()> {
     Ok(())
 }
 
-fn user_account_session(cred: Credentials) -> errors::Result<()> {
-    let user_session = utils::user_session_with_cached_refresh_token(&cred)?;
+async fn user_account_session(cred: Credentials) -> errors::Result<()> {
+    let user_session = utils::user_session_with_cached_refresh_token(&cred).await?;
 
     assert_eq!(user_session.user_id, utils::TEST_USER_ID);
     assert_eq!(user_session.project_id(), cred.project_id);
 
     println!("user::Session::by_access_token");
-    let user_session = sessions::user::Session::by_access_token(&cred, &user_session.access_token_unchecked())?;
+    let user_session =
+        sessions::user::Session::by_access_token(&cred, &user_session.access_token_unchecked().await).await?;
 
     assert_eq!(user_session.user_id, utils::TEST_USER_ID);
 
@@ -117,13 +121,14 @@ fn user_account_session(cred: Credentials) -> errors::Result<()> {
             Some(doc_id),
             &obj,
             documents::WriteOptions::default(),
-        )?,
+        )
+        .await?,
         doc_id,
     );
 
     // Test reading
     println!("user::Session documents::read");
-    let read: DemoDTO = documents::read(&user_session, "tests", doc_id)?;
+    let read: DemoDTO = documents::read(&user_session, "tests", doc_id).await?;
 
     assert_eq!(read.a_string, "abc");
     assert_eq!(read.an_int, 12);
@@ -135,22 +140,25 @@ fn user_account_session(cred: Credentials) -> errors::Result<()> {
         "abc".into(),
         dto::FieldOperator::EQUAL,
         "a_string",
-    )?
+    )
+    .await?
     .collect();
     assert_eq!(results.len(), 1);
-    let doc: DemoDTO = documents::read_by_name(&user_session, &results.get(0).unwrap().name)?;
+    let doc: DemoDTO = documents::read_by_name(&user_session, &results.get(0).unwrap().name).await?;
     assert_eq!(doc.a_string, "abc");
 
     let mut count = 0;
-    let list_it: documents::List<DemoDTO, _> = documents::list(&user_session, "tests".to_owned());
+    let list_it = documents::list(&user_session, "tests".to_owned())
+        .collect::<Vec<errors::Result<(DemoDTO, _)>>>()
+        .await;
     for _doc in list_it {
         count += 1;
     }
     assert_eq!(count, 2);
 
     // test if the call fails for a non existing document
     println!("user::Session documents::delete");
-    let r = documents::delete(&user_session, "tests/non_existing", true);
+    let r = documents::delete(&user_session, "tests/non_existing", true).await;
     assert!(r.is_err());
     match r.err().unwrap() {
         errors::FirebaseError::APIError(code, message, context) => {
@@ -161,7 +169,7 @@ fn user_account_session(cred: Credentials) -> errors::Result<()> {
         _ => panic!("Expected an APIError"),
     };
 
-    documents::delete(&user_session, &("tests/".to_owned() + doc_id), false)?;
+    documents::delete(&user_session, &("tests/".to_owned() + doc_id), false).await?;
 
     // Check if document is indeed removed
     println!("user::Session documents::query");
@@ -171,71 +179,76 @@ fn user_account_session(cred: Credentials) -> errors::Result<()> {
         "abc".into(),
         dto::FieldOperator::EQUAL,
         "a_string",
-    )?
+    )
+    .await?
     .count();
     assert_eq!(count, 0);
 
     println!("user::Session documents::query for f64");
     let f: f64 = 13.37;
-    let count = documents::query(&user_session, "tests", f.into(), dto::FieldOperator::EQUAL, "a_float")?.count();
+
+    let count = documents::query(&user_session, "tests", f.into(), dto::FieldOperator::EQUAL, "a_float").await?;
+
+    let count = count.count();
     assert_eq!(count, 0);
 
     Ok(())
 }
 
-fn main() -> errors::Result<()> {
+#[tokio::main]
+async fn main() -> errors::Result<()> {
     // Search for a credentials file in the root directory
     use std::path::PathBuf;
     let mut credential_file = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
     credential_file.push("firebase-service-account.json");
-    let mut cred = Credentials::from_file(credential_file.to_str().unwrap())?;
+    let cred = Credentials::from_file(credential_file.to_str().unwrap()).await?;
 
     // Only download the public keys once, and cache them.
-    let jwkset = utils::from_cache_file(credential_file.with_file_name("cached_jwks.jwks").as_path(), &cred)?;
+    let jwkset = utils::from_cache_file(credential_file.with_file_name("cached_jwks.jwks").as_path(), &cred).await?;
     cred.add_jwks_public_keys(&jwkset);
-    cred.verify()?;
+    cred.verify().await?;
 
     // Perform some db operations via a service account session
-    service_account_session(cred.clone())?;
+    service_account_session(cred.clone()).await?;
 
     // Perform some db operations via a firebase user session
-    user_account_session(cred)?;
+    user_account_session(cred).await?;
 
     Ok(())
 }
 
 /// For integration tests and doc code snippets: Create a Credentials instance.
 /// Necessary public jwk sets are downloaded or re-used if already present.
 #[cfg(test)]
-fn valid_test_credentials() -> errors::Result<Credentials> {
+async fn valid_test_credentials() -> errors::Result<Credentials> {
     use std::path::PathBuf;
     let mut jwks_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
     jwks_path.push("firebase-service-account.jwks");
 
-    let mut cred: Credentials = Credentials::new(include_str!("../firebase-service-account.json"))?;
+    let mut cred: Credentials = Credentials::new(include_str!("../tests/service-account-test.json"))?;
 
     // Only download the public keys once, and cache them.
-    let jwkset = utils::from_cache_file(jwks_path.as_path(), &cred)?;
+    let jwkset = utils::from_cache_file(jwks_path.as_path(), &cred).await?;
     cred.add_jwks_public_keys(&jwkset);
     cred.verify()?;
 
     Ok(cred)
 }
 
-#[test]
-fn valid_test_credentials_test() -> errors::Result<()> {
-    valid_test_credentials()?;
+#[tokio::test]
+async fn valid_test_credentials_test() -> errors::Result<()> {
+    valid_test_credentials().await?;
     Ok(())
 }
 
-#[test]
-fn service_account_session_test() -> errors::Result<()> {
-    service_account_session(valid_test_credentials()?)?;
+#[tokio::test]
+async fn service_account_session_test() -> errors::Result<()> {
+    service_account_session(valid_test_credentials().await?).await?;
     Ok(())
 }
 
-#[test]
-fn user_account_session_test() -> errors::Result<()> {
-    user_account_session(valid_test_credentials()?)?;
+#[tokio::test]
+async fn user_account_session_test() -> errors::Result<()> {
+    user_account_session(valid_test_credentials().await?).await?;
     Ok(())
 }

examples/firebase_user.rs

@@ -3,13 +3,16 @@ use firestore_db_and_auth::*;
 
 const TEST_USER_ID: &str = include_str!("test_user_id.txt");
 
-fn main() -> errors::Result<()> {
-    let cred = Credentials::from_file("firebase-service-account.json").expect("Read credentials file");
+#[tokio::main]
+async fn main() -> errors::Result<()> {
+    let cred = Credentials::from_file("firebase-service-account.json")
+        .await
+        .expect("Read credentials file");
 
-    let user_session = UserSession::by_user_id(&cred, TEST_USER_ID, false)?;
+    let user_session = UserSession::by_user_id(&cred, TEST_USER_ID, false).await?;
 
     println!("users::user_info");
-    let user_info_container = users::user_info(&user_session)?;
+    let user_info_container = users::user_info(&user_session).await?;
     assert_eq!(user_info_container.users[0].localId.as_ref().unwrap(), TEST_USER_ID);
 
     Ok(())

examples/own_auth.rs

@@ -1,41 +1,37 @@
-use firestore_db_and_auth::errors::FirebaseError::APIError;
 use firestore_db_and_auth::{documents, errors, Credentials, FirebaseAuthBearer};
 
 /// Define your own structure that will implement the FirebaseAuthBearer trait
 struct MyOwnSession {
     /// The google credentials
     pub credentials: Credentials,
-    pub blocking_client: reqwest::blocking::Client,
     pub client: reqwest::Client,
     access_token: String,
 }
 
+#[async_trait::async_trait]
 impl FirebaseAuthBearer for MyOwnSession {
     fn project_id(&self) -> &str {
         &self.credentials.project_id
     }
     /// An access token. If a refresh token is known and the access token expired,
     /// the implementation should try to refresh the access token before returning.
-    fn access_token(&self) -> String {
+    async fn access_token(&self) -> String {
         self.access_token.clone()
     }
     /// The access token, unchecked. Might be expired or in other ways invalid.
-    fn access_token_unchecked(&self) -> String {
+    async fn access_token_unchecked(&self) -> String {
         self.access_token.clone()
     }
     /// The reqwest http client.
     /// The `Client` holds a connection pool internally, so it is advised that it is reused for multiple, successive connections.
-    fn client(&self) -> &reqwest::blocking::Client {
-        &self.blocking_client
-    }
-
-    fn client_async(&self) -> &reqwest::Client {
+    fn client(&self) -> &reqwest::Client {
         &self.client
     }
 }
 
-fn main() -> errors::Result<()> {
-    let credentials = Credentials::from_file("firebase-service-account.json")?;
+#[tokio::main]
+async fn main() -> errors::Result<()> {
+    let credentials = Credentials::from_file("firebase-service-account.json").await?;
     #[derive(serde::Serialize)]
     struct TestData {
         an_int: u32,
@@ -44,7 +40,6 @@ fn main() -> errors::Result<()> {
 
     let session = MyOwnSession {
         credentials,
-        blocking_client: reqwest::blocking::Client::new(),
         client: reqwest::Client::new(),
         access_token: "The access token".to_owned(),
     };
@@ -56,12 +51,13 @@ fn main() -> errors::Result<()> {
         Some("test_doc"),
         &t,
         documents::WriteOptions::default(),
-    )?;
+    )
+    .await?;
     Ok(())
 }
 
-#[test]
-fn own_auth_test() {
+#[tokio::test]
+async fn own_auth_test() {
     if let Err(APIError(code, str_code, context)) = main() {
         assert_eq!(str_code, "Request had invalid authentication credentials. Expected OAuth 2 access token, login cookie or other valid authentication credential. See https://developers.google.com/identity/sign-in/web/devconsole-project.");
         assert_eq!(context, "test_doc");

examples/session_cookie.rs

@@ -4,22 +4,23 @@ use chrono::Duration;
 
 mod utils;
 
-fn main() -> Result<(), FirebaseError> {
+#[tokio::main]
+async fn main() -> Result<(), FirebaseError> {
     // Search for a credentials file in the root directory
     use std::path::PathBuf;
 
     let mut credential_file = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
     credential_file.push("firebase-service-account.json");
-    let mut cred = Credentials::from_file(credential_file.to_str().unwrap())?;
+    let cred = Credentials::from_file(credential_file.to_str().unwrap()).await?;
 
     // Only download the public keys once, and cache them.
-    let jwkset = utils::from_cache_file(credential_file.with_file_name("cached_jwks.jwks").as_path(), &cred)?;
+    let jwkset = utils::from_cache_file(credential_file.with_file_name("cached_jwks.jwks").as_path(), &cred).await?;
     cred.add_jwks_public_keys(&jwkset);
-    cred.verify()?;
+    cred.verify().await?;
 
-    let user_session = utils::user_session_with_cached_refresh_token(&cred)?;
+    let user_session = utils::user_session_with_cached_refresh_token(&cred).await?;
 
-    let cookie = session_cookie::create(&cred, user_session.access_token(), Duration::seconds(3600))?;
+    let cookie = session_cookie::create(&cred, user_session.access_token().await, Duration::seconds(3600)).await?;
     println!("Created session cookie: {}", cookie);
 
     Ok(())