Upgraded tokio, warp, hyper and actix (#912)

* Upgraded tokio, warp, hyper and actix

* Code formatting

* actix-web temporary version fix specification

* Error handling fix on juniper_rocket

Co-authored-by: Christian Legnitto <LegNeato@users.noreply.github.com>

Author: engylemure

docs/book/content/advanced/subscriptions.md

@@ -51,7 +51,7 @@ type StringStream = Pin<Box<dyn Stream<Item = Result<String, FieldError>> + Send
 #[graphql_subscription(context = Database)]
 impl Subscription {
     async fn hello_world() -> StringStream {
-        let stream = tokio::stream::iter(vec![
+        let stream = futures::stream::iter(vec![
             Ok(String::from("Hello")),
             Ok(String::from("World!"))
         ]);
@@ -123,7 +123,7 @@ where [`Connection`][Connection] is a `Stream` of values returned by the operati
 # impl Subscription {
 #     async fn hello_world() -> StringStream {
 #         let stream =
-#             tokio::stream::iter(vec![Ok(String::from("Hello")), Ok(String::from("World!"))]);
+#             futures::stream::iter(vec![Ok(String::from("Hello")), Ok(String::from("World!"))]);
 #         Box::pin(stream)
 #     }
 # }

docs/book/tests/Cargo.toml

@@ -16,7 +16,7 @@ iron = "0.5"
 mount = "0.4"
 skeptic = "0.13"
 serde_json = "1.0"
-tokio = { version = "0.2", features = ["blocking", "macros", "rt-core", "rt-util", "stream"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 uuid = "0.8"
 
 [build-dependencies]

examples/actix_subscriptions/Cargo.toml

@@ -6,15 +6,15 @@ authors = ["Mihai Dinculescu <mihai.dinculescu@outlook.com>"]
 publish = false
 
 [dependencies]
-actix-web = "3.3"
-actix-cors = "0.5"
+actix-web = "4.0.0-beta.5"
+actix-cors = "0.6.0-beta.1"
 futures = "0.3"
-tokio = { version = "0.2", features = ["macros", "rt-core"] }
 env_logger = "0.8"
 serde = "1.0"
 serde_json = "1.0"
-rand = "0.7"
-
+rand = "0.8"
+tokio = "1.0"
+async-stream = "0.3"
 juniper = { path = "../../juniper", features = ["expose-test-schema"] }
 juniper_actix = { path = "../../juniper_actix", features = ["subscriptions"] }
 juniper_graphql_ws = { path = "../../juniper_graphql_ws" }

examples/actix_subscriptions/src/main.rs

@@ -1,7 +1,7 @@
 use std::{env, pin::Pin, time::Duration};
 
 use actix_cors::Cors;
-use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer};
+use actix_web::{http::header, middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer};
 
 use juniper::{
     graphql_object, graphql_subscription,
@@ -64,27 +64,29 @@ impl Subscription {
 
         use rand::{rngs::StdRng, Rng, SeedableRng};
         let mut rng = StdRng::from_entropy();
-
-        let stream = tokio::time::interval(Duration::from_secs(3)).map(move |_| {
+        let mut interval = tokio::time::interval(Duration::from_secs(5));
+        let stream = async_stream::stream! {
             counter += 1;
-
-            if counter == 2 {
-                Err(FieldError::new(
-                    "some field error from handler",
-                    Value::Scalar(DefaultScalarValue::String(
-                        "some additional string".to_string(),
-                    )),
-                ))
-            } else {
-                let random_id = rng.gen_range(1000, 1005).to_string();
-                let human = context.get_human(&random_id).unwrap().clone();
-
-                Ok(RandomHuman {
-                    id: human.id().to_owned(),
-                    name: human.name().unwrap().to_owned(),
-                })
+            loop {
+                interval.tick().await;
+                if counter == 2 {
+                    yield Err(FieldError::new(
+                        "some field error from handler",
+                        Value::Scalar(DefaultScalarValue::String(
+                            "some additional string".to_string(),
+                        )),
+                    ))
+                } else {
+                    let random_id = rng.gen_range(1000..1005).to_string();
+                    let human = context.get_human(&random_id).unwrap().clone();
+
+                    yield Ok(RandomHuman {
+                        id: human.id().to_owned(),
+                        name: human.name().unwrap().to_owned(),
+                    })
+                }
             }
-        });
+        };
 
         Box::pin(stream)
     }
@@ -117,7 +119,10 @@ async fn main() -> std::io::Result<()> {
             .wrap(middleware::Logger::default())
             .wrap(
                 Cors::default()
+                    .allowed_origin("http://127.0.0.1:8080")
                     .allowed_methods(vec!["POST", "GET"])
+                    .allowed_headers(vec![header::AUTHORIZATION, header::ACCEPT])
+                    .allowed_header(header::CONTENT_TYPE)
                     .supports_credentials()
                     .max_age(3600),
             )
@@ -130,7 +135,7 @@ async fn main() -> std::io::Result<()> {
             .service(web::resource("/playground").route(web::get().to(playground)))
             .default_service(web::route().to(|| {
                 HttpResponse::Found()
-                    .header("location", "/playground")
+                    .append_header((header::LOCATION, "/playground"))
                     .finish()
             }))
     })

examples/basic_subscriptions/Cargo.toml

@@ -11,7 +11,7 @@ authors = ["Jordao Rosario <jordao.rosario01@gmail.com>"]
 futures = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tokio = { version = "0.2", features = ["rt-core", "macros", "stream"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 
 juniper = { path = "../../juniper" }
 juniper_subscriptions = { path = "../../juniper_subscriptions" }

examples/basic_subscriptions/src/main.rs

@@ -37,7 +37,7 @@ type StringStream = Pin<Box<dyn Stream<Item = Result<String, FieldError>> + Send
 impl Subscription {
     async fn hello_world() -> StringStream {
         let stream =
-            tokio::stream::iter(vec![Ok(String::from("Hello")), Ok(String::from("World!"))]);
+            futures::stream::iter(vec![Ok(String::from("Hello")), Ok(String::from("World!"))]);
         Box::pin(stream)
     }
 }

examples/warp_async/Cargo.toml

@@ -13,5 +13,5 @@ env_logger = "0.8.1"
 futures = "0.3.1"
 log = "0.4.8"
 reqwest = { version = "0.11", features = ["rustls-tls"] }
-tokio = { version = "0.2", features = ["rt-core", "macros"] }
-warp = "0.2"
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+warp = "0.3"

examples/warp_subscriptions/Cargo.toml

@@ -10,9 +10,9 @@ futures = "0.3.1"
 log = "0.4.8"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tokio = { version = "0.2", features = ["rt-core", "macros"] }
-warp = "0.2.1"
-
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+warp = "0.3"
+async-stream = "0.3"
 juniper = { path = "../../juniper" }
 juniper_graphql_ws = { path = "../../juniper_graphql_ws" }
 juniper_warp = { path = "../../juniper_warp", features = ["subscriptions"] }

examples/warp_subscriptions/src/main.rs

@@ -109,23 +109,27 @@ struct Subscription;
 impl Subscription {
     async fn users() -> UsersStream {
         let mut counter = 0;
-        let stream = tokio::time::interval(Duration::from_secs(5)).map(move |_| {
+        let mut interval = tokio::time::interval(Duration::from_secs(5));
+        let stream = async_stream::stream! {
             counter += 1;
-            if counter == 2 {
-                Err(FieldError::new(
-                    "some field error from handler",
-                    Value::Scalar(DefaultScalarValue::String(
-                        "some additional string".to_string(),
-                    )),
-                ))
-            } else {
-                Ok(User {
-                    id: counter,
-                    kind: UserKind::Admin,
-                    name: "stream user".to_string(),
-                })
+            loop {
+                interval.tick().await;
+                if counter == 2 {
+                    yield Err(FieldError::new(
+                        "some field error from handler",
+                        Value::Scalar(DefaultScalarValue::String(
+                            "some additional string".to_string(),
+                        )),
+                    ))
+                } else {
+                    yield Ok(User {
+                        id: counter,
+                        kind: UserKind::Admin,
+                        name: "stream user".to_string(),
+                    })
+                }
             }
-        });
+        };
 
         Box::pin(stream)
     }

integration_tests/async_await/Cargo.toml

@@ -8,4 +8,4 @@ publish = false
 [dependencies]
 juniper = { path = "../../juniper" }
 futures = "0.3.1"
-tokio = { version = "0.2", features = ["rt-core", "time", "macros"] }
+tokio = { version = "1", features = ["rt", "time", "macros"] }