Merge branch 'tim/maint' into 'master'

Upgrade to Tokio 1.3

See merge request TankerHQ/sdk-rust!26

Author: tux3

Cargo.toml

@@ -11,16 +11,18 @@ include = ["native", "src", "tests", "build.rs"]
 futures = "0.3"
 num_enum = "0.5"
 lazy_static = "1.4"
-tokio = { version = "0.2", features = ["sync"] }
+tokio = { version = "1.3", features = ["sync"] }
+# Tokio's mpsc::sync channel has a bug in try_recv(), in the meantime we use async_channel
+async-channel = "1.6.1"
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["macros", "rt-threaded"] }
-reqwest = { version = "0.10.8", features = ["json"] }
+tokio = { version = "1.3", features = ["macros", "rt-multi-thread"] }
+reqwest = { version = "0.11", features = ["json"] }
 openssl = { version = "0.10", features = ["vendored"] }
 serde_json = "1.0"
 double-checked-cell-async = "2.0.2"
 rand = "0.7.3"
-base64 = "0.12"
+base64 = "0.13"
 variant_count = "1.0"
 blake2 = "0.9"
 ed25519-dalek = "1.0"

src/ctanker/cstream.rs

@@ -25,6 +25,7 @@ use crate::ctanker::*;
 use crate::error::Error;
 
 use ::core::pin::Pin;
+use async_channel::{bounded, Receiver, Sender, TryRecvError};
 use futures::executor::block_on;
 use futures::future::{select, Either};
 use futures::io::{AsyncRead, AsyncReadExt};
@@ -34,8 +35,6 @@ use futures::FutureExt;
 use std::cmp::min;
 use std::future::Future;
 use std::sync::Mutex;
-use tokio::sync::mpsc::error::TryRecvError;
-use tokio::sync::mpsc::{channel, Receiver, Sender};
 
 #[derive(Debug, Clone)]
 struct ReadOperation {
@@ -64,7 +63,7 @@ struct TankerStream<UserStream: AsyncRead + Unpin> {
 
 impl<UserStream: AsyncRead + Unpin> TankerStream<UserStream> {
     fn new() -> Self {
-        let (sender, receiver) = channel(1);
+        let (sender, receiver) = bounded(1);
         TankerStream {
             user_stream: None,
             tanker_stream_handle: std::ptr::null_mut(),
@@ -184,10 +183,10 @@ impl<UserStream: AsyncRead + Unpin> AsyncRead for TankerStream<UserStream> {
                     );
                     self.read_operation = Some(read_operation);
                 }
-                Err(TryRecvError::Empty) => {} // No work to do
-                Err(e) => {
-                    panic!("error reading channel: {}", e);
+                Err(TryRecvError::Closed) => {
+                    panic!("error reading channel: closed");
                 }
+                Err(TryRecvError::Empty) => {} // Channel still open, but no message
             }
 
             // Process the ReadOperation if there is one in progress

tests/encryption_session_tests.rs

@@ -4,7 +4,7 @@ use futures::AsyncReadExt;
 use identity::TestApp;
 use tankersdk::*;
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn open_close_enc_sess() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -16,7 +16,7 @@ async fn open_close_enc_sess() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn share_with_enc_sess() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;
@@ -32,7 +32,7 @@ async fn share_with_enc_sess() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_stream_with_enc_sess() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;
@@ -52,7 +52,7 @@ async fn encrypt_stream_with_enc_sess() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn resource_id_of_enc_sess_matches_ciphertext() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -67,7 +67,7 @@ async fn resource_id_of_enc_sess_matches_ciphertext() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn resource_id_of_different_enc_sess_are_different() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;

tests/group_tests.rs

@@ -4,7 +4,7 @@ use identity::TestApp;
 use std::iter;
 use tankersdk::*;
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn cannot_create_empty_group() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -20,7 +20,7 @@ async fn cannot_create_empty_group() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn create_valid_group() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;
@@ -36,7 +36,7 @@ async fn create_valid_group() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_and_share_with_external_group() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice_id = app.create_identity(None);
@@ -58,7 +58,7 @@ async fn encrypt_and_share_with_external_group() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn share_with_external_group() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice_id = app.create_identity(None);
@@ -83,7 +83,7 @@ async fn share_with_external_group() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn add_member_to_group() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice_id = app.create_identity(None);

tests/stream_tests.rs

@@ -8,7 +8,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use tankersdk::{Error, ErrorCode};
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_stream_and_decrypt() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -26,7 +26,7 @@ async fn encrypt_stream_and_decrypt() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_and_decrypt_stream() -> Result<(), futures::io::Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -42,7 +42,7 @@ async fn encrypt_and_decrypt_stream() -> Result<(), futures::io::Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_stream_and_decrypt_stream() -> Result<(), futures::io::Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -84,7 +84,7 @@ impl AsyncRead for ErrorAfter {
     }
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_stream_with_error() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -104,7 +104,7 @@ async fn encrypt_stream_with_error() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn decrypt_stream_with_early_error() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -123,7 +123,7 @@ async fn decrypt_stream_with_early_error() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn decrypt_stream_with_tanker_error() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;

tests/tanker_tests.rs

@@ -13,7 +13,7 @@ fn core_native_version() {
     assert!(!Core::native_version().is_empty())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn tanker_create() -> Result<(), Error> {
     let app = TestApp::get().await;
     let opts = Options::new(app.id().to_owned(), ":memory:".to_string())
@@ -23,7 +23,7 @@ async fn tanker_create() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn tanker_bad_create() {
     let opts = Options::new("bad-app-id".to_string(), ":memory:".to_string())
         .with_sdk_type("sdk-rust-test".to_string());
@@ -33,7 +33,7 @@ async fn tanker_bad_create() {
     assert_eq!(err.code(), ErrorCode::InvalidArgument);
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn start_stop_session() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = Core::new(app.make_options()).await?;
@@ -49,7 +49,7 @@ async fn start_stop_session() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn self_revoke() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -62,7 +62,7 @@ async fn self_revoke() -> Result<(), Error> {
     tanker.stop().await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn has_correct_device_list() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -75,7 +75,7 @@ async fn has_correct_device_list() -> Result<(), Error> {
     tanker.stop().await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_and_decrypt() -> Result<(), Error> {
     let app = TestApp::get().await;
     let tanker = app.start_anonymous(&app.create_identity(None)).await?;
@@ -89,7 +89,7 @@ async fn encrypt_and_decrypt() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn share_then_decrypt() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;
@@ -112,7 +112,7 @@ async fn share_then_decrypt() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_and_share_then_decrypt() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;
@@ -131,7 +131,7 @@ async fn encrypt_and_share_then_decrypt() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn encrypt_no_share_with_self() -> Result<(), Error> {
     let app = TestApp::get().await;
     let alice = app.start_anonymous(&app.create_identity(None)).await?;
@@ -155,7 +155,7 @@ async fn encrypt_no_share_with_self() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn share_with_provisional_user() -> Result<(), Error> {
     let message = b"Variable 'message' is never used";
     let app = TestApp::get().await;
@@ -190,7 +190,7 @@ async fn share_with_provisional_user() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn attach_provisional_with_single_verif() -> Result<(), Error> {
     let message = b"Variable 'message' is never used";
     let app = TestApp::get().await;
@@ -223,14 +223,14 @@ async fn attach_provisional_with_single_verif() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn prehash_password_empty() -> Result<(), Error> {
     let err = Core::prehash_password("").unwrap_err();
     assert_eq!(err.code(), ErrorCode::InvalidArgument);
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn prehash_password_test_vector_1() -> Result<(), Error> {
     let input = "super secretive password";
     let expected = "UYNRgDLSClFWKsJ7dl9uPJjhpIoEzadksv/Mf44gSHI=";
@@ -239,7 +239,7 @@ async fn prehash_password_test_vector_1() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn prehash_password_test_vector_2() -> Result<(), Error> {
     let input = "test éå 한국어 😃";
     let expected = "Pkn/pjub2uwkBDpt2HUieWOXP5xLn0Zlen16ID4C7jI=";

tests/verify_tests.rs

@@ -4,7 +4,7 @@ use identity::TestApp;
 use serde_json::{json, Value};
 use tankersdk::*;
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn validate_new_device_with_verif_key() -> Result<(), Error> {
     let app = TestApp::get().await;
     let id = &app.create_identity(None);
@@ -23,7 +23,7 @@ async fn validate_new_device_with_verif_key() -> Result<(), Error> {
     tanker.stop().await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn setup_and_use_passphrase() -> Result<(), Error> {
     let app = TestApp::get().await;
     let id = &app.create_identity(None);
@@ -41,7 +41,7 @@ async fn setup_and_use_passphrase() -> Result<(), Error> {
     tanker.stop().await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn unlock_with_updated_passphrase() -> Result<(), Error> {
     let app = TestApp::get().await;
     let id = &app.create_identity(None);
@@ -61,7 +61,7 @@ async fn unlock_with_updated_passphrase() -> Result<(), Error> {
     tanker.stop().await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn check_passphrase_is_setup() -> Result<(), Error> {
     let app = TestApp::get().await;
     let id = &app.create_identity(None);
@@ -77,7 +77,7 @@ async fn check_passphrase_is_setup() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn check_email_verif_is_setup() -> Result<(), Error> {
     let app = TestApp::get().await;
     let id = &app.create_identity(None);
@@ -97,7 +97,7 @@ async fn check_email_verif_is_setup() -> Result<(), Error> {
     Ok(())
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn unlock_with_verif_code() -> Result<(), Error> {
     let app = TestApp::get().await;
     let id = &app.create_identity(None);
@@ -123,7 +123,7 @@ async fn unlock_with_verif_code() -> Result<(), Error> {
     tanker.stop().await
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn unlock_with_oidc_id_token() -> Result<(), Box<dyn std::error::Error>> {
     let app = TestApp::get().await;
     let oidc = app.get_oidc_config();