anna
/
async-graphql
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Replace futures with futures_util

This commit is contained in:
Koxiaet 2020-10-16 07:49:22 +01:00
parent 6f47b11fed
commit 12fe7445c7
26 changed files with 95 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
Cargo.toml
View File

@ -14,10 +14,23 @@ categories = ["network-programming", "asynchronous"]
readme = "README.md"
[features]
default = ["apollo_tracing", "apollo_persisted_queries", "uuid", "bson", "chrono", "chrono-tz", "log", "multipart", "tracing", "url", "unblock", "string_number"]
default = [
"apollo_persisted_queries",
"apollo_tracing",
"bson",
"chrono",
"chrono-tz",
"log",
"multipart",
"string_number",
"tracing",
"unblock",
"url",
"uuid",
]
apollo_tracing = ["chrono"]
apollo_persisted_queries = ["lru"]
multipart = ["multer", "bytes", "tempfile"]
apollo_persisted_queries = ["async-mutex", "lru"]
multipart = ["bytes", "multer", "tempfile"]
unblock = ["blocking"]
string_number = ["num-traits"]
# Used for doc(cfg())
@ -31,7 +44,7 @@ async-graphql-parser = { path = "parser", version = "=2.0.3" }
async-stream = "0.3"
async-trait = "0.1.41"
fnv = "1.0.6"
futures = "0.3.6"
futures-util = { version = "0.3.6", default-features = false, features = ["io"] }
indexmap = "1.6.0"
once_cell = "1.3.1"
pin-project-lite = "0.1.10"
@ -43,23 +56,26 @@ thiserror = "1.0.21"
static_assertions = "1.1.0"
# Feature optional dependencies
uuid = { version = "0.8.1", optional = true, features = ["v4", "serde"] }
bson = { version = "1.0.0", optional = true }
chrono = { version = "0.4.15", optional = true }
chrono-tz = { version = "0.5.1", optional = true }
log = { version = "0.4.11", optional = true }
tracing = { version = "0.1.21", optional = true }
url = { version = "2.1.1", optional = true }
num-traits = { version = "0.2.12", optional = true }
lru = { version = "0.6.0", optional = true }
uuid = { version = "0.8.1", optional = true, features = ["v4", "serde"] }
bytes = { version = "0.5.4", optional = true }
multer = { version = "1.2.2", optional = true }
tempfile = { version = "3.1.0", optional = true }
# Non-feature optional dependencies
async-mutex = { version = "1.4.0", optional = true }
blocking = { version = "1.0.0", optional = true }
bytes = { version = "0.5.4", optional = true }
lru = { version = "0.6.0", optional = true }
multer = { version = "1.2.2", optional = true }
num-traits = { version = "0.2.12", optional = true }
tempfile = { version = "3.1.0", optional = true }
[dev-dependencies]
async-std = { version = "1.6.5", features = ["attributes"] }
async-channel = "1.5.1"
[package.metadata.docs.rs]
features = ["nightly"]

2
derive/src/merged_subscription.rs
View File

@ -81,7 +81,7 @@ pub fn generate(object_args: &args::MergedSubscription) -> GeneratorResult<Token
fn create_field_stream<'__life>(
&'__life self,
ctx: &'__life #crate_name::Context<'__life>
) -> Option<::std::pin::Pin<::std::boxed::Box<dyn #crate_name::futures::Stream<Item = #crate_name::ServerResult<#crate_name::Value>> + ::std::marker::Send + '__life>>> {
) -> Option<::std::pin::Pin<::std::boxed::Box<dyn #crate_name::futures_util::stream::Stream<Item = #crate_name::ServerResult<#crate_name::Value>> + ::std::marker::Send + '__life>>> {
None #create_field_stream
}
}

18
derive/src/subscription.rs
View File

@ -247,7 +247,7 @@ pub fn generate(
#(#schema_args)*
args
},
ty: <<#stream_ty as #crate_name::futures::stream::Stream>::Item as #crate_name::Type>::create_type_info(registry),
ty: <<#stream_ty as #crate_name::futures_util::stream::Stream>::Item as #crate_name::Type>::create_type_info(registry),
deprecation: #field_deprecation,
cache_control: Default::default(),
external: false,
@ -281,7 +281,7 @@ pub fn generate(
let pos = ctx.item.pos;
let schema_env = ctx.schema_env.clone();
let query_env = ctx.query_env.clone();
let stream = #crate_name::futures::StreamExt::then(#create_field_stream, {
let stream = #crate_name::futures_util::stream::StreamExt::then(#create_field_stream, {
let field_name = field_name.clone();
move |msg| {
let schema_env = schema_env.clone();
@ -316,7 +316,7 @@ pub fn generate(
resolve_id,
path_node: ctx_selection_set.path_node.as_ref().unwrap(),
parent_type: #gql_typename,
return_type: &<<#stream_ty as #crate_name::futures::stream::Stream>::Item as #crate_name::Type>::qualified_type_name(),
return_type: &<<#stream_ty as #crate_name::futures_util::stream::Stream>::Item as #crate_name::Type>::qualified_type_name(),
};
query_env.extensions.resolve_start(&ctx_extension, &ri);
@ -330,17 +330,17 @@ pub fn generate(
}
}
});
#crate_name::ServerResult::Ok(#crate_name::futures::StreamExt::scan(
#crate_name::ServerResult::Ok(#crate_name::futures_util::stream::StreamExt::scan(
stream,
false,
|errored, item| {
if *errored {
return #crate_name::futures::future::ready(None);
return #crate_name::futures_util::future::ready(None);
}
if item.is_err() {
*errored = true;
}
#crate_name::futures::future::ready(Some(item))
#crate_name::futures_util::future::ready(Some(item))
},
))
};
@ -349,8 +349,8 @@ pub fn generate(
#(#cfg_attrs)*
if ctx.item.node.name.node == #field_name {
return ::std::option::Option::Some(::std::boxed::Box::pin(
#crate_name::futures::TryStreamExt::try_flatten(
#crate_name::futures::stream::once((move || async move { #stream_fn })())
#crate_name::futures_util::stream::TryStreamExt::try_flatten(
#crate_name::futures_util::stream::once((move || async move { #stream_fn })())
)
));
}
@ -392,7 +392,7 @@ pub fn generate(
fn create_field_stream<'__life>(
&'__life self,
ctx: &'__life #crate_name::Context<'__life>,
) -> ::std::option::Option<::std::pin::Pin<::std::boxed::Box<dyn #crate_name::futures::Stream<Item = #crate_name::ServerResult<#crate_name::Value>> + Send + '__life>>> {
) -> ::std::option::Option<::std::pin::Pin<::std::boxed::Box<dyn #crate_name::futures_util::stream::Stream<Item = #crate_name::ServerResult<#crate_name::Value>> + Send + '__life>>> {
#(#create_stream)*
None
}

2
integrations/warp/src/subscription.rs
View File

@ -11,7 +11,7 @@ use warp::{Filter, Rejection, Reply};
/// use async_graphql::*;
/// use async_graphql_warp::*;
/// use warp::Filter;
/// use futures::{Stream, StreamExt};
/// use futures_util::stream::{Stream, StreamExt};
/// use std::time::Duration;
///
/// struct QueryRoot;

2
src/context.rs
View File

@ -299,7 +299,7 @@ impl Display for ResolveId {
/// **This type is not stable and should not be used directly.**
#[derive(Clone)]
pub struct ContextBase<'a, T> {
#[allow(missing_docs)]
/// The current path node being resolved.
pub path_node: Option<QueryPathNode<'a>>,
pub(crate) resolve_id: ResolveId,
pub(crate) inc_resolve_id: &'a AtomicUsize,

2
src/extensions/apollo_persisted_queries.rs
View File

@ -2,7 +2,7 @@
use std::sync::Arc;
use futures::lock::Mutex;
use async_mutex::Mutex;
use serde::Deserialize;
use crate::extensions::{Extension, ExtensionContext, ExtensionFactory};

5
src/http/mod.rs
View File

@ -6,8 +6,7 @@ mod multipart;
mod playground_source;
mod websocket;
use futures::io::AsyncRead;
use futures::AsyncReadExt;
use futures_util::io::{AsyncRead, AsyncReadExt};
use crate::{BatchRequest, ParseRequestError, Request};
@ -59,7 +58,7 @@ pub async fn receive_json(body: impl AsyncRead) -> Result<Request, ParseRequestE
/// Receive a GraphQL batch request from a body as JSON.
pub async fn receive_batch_json(body: impl AsyncRead) -> Result<BatchRequest, ParseRequestError> {
let mut data = Vec::new();
futures::pin_mut!(body);
futures_util::pin_mut!(body);
body.read_to_end(&mut data)
.await
.map_err(ParseRequestError::Io)?;

6
src/http/multipart.rs
View File

@ -4,8 +4,8 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::Bytes;
use futures::io::AsyncRead;
use futures::stream::Stream;
use futures_util::io::AsyncRead;
use futures_util::stream::Stream;
use multer::{Constraints, Multipart, SizeLimit};
use pin_project_lite::pin_project;
@ -161,7 +161,7 @@ impl<T: AsyncRead> Stream for ReaderStream<T> {
let this = self.project();
Poll::Ready(
match futures::ready!(this.reader.poll_read(cx, this.buf)?) {
match futures_util::ready!(this.reader.poll_read(cx, this.buf)?) {
0 => None,
size => Some(Ok(Bytes::copy_from_slice(&this.buf[..size]))),
},

2
src/http/websocket.rs
View File

@ -5,7 +5,7 @@ use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use futures::Stream;
use futures_util::stream::Stream;
use pin_project_lite::pin_project;
use serde::{Deserialize, Serialize};

12
src/lib.rs
View File

@ -147,7 +147,7 @@ pub use async_trait;
#[doc(hidden)]
pub use context::ContextSelectionSet;
#[doc(hidden)]
pub use futures;
pub use futures_util;
#[doc(hidden)]
pub use indexmap;
#[doc(hidden)]
@ -712,7 +712,7 @@ pub use async_graphql_derive::Union;
///
/// ```rust
/// use async_graphql::*;
/// use futures::{Stream, StreamExt};
/// use futures_util::stream::{Stream, StreamExt};
///
/// struct SubscriptionRoot;
///
@ -720,7 +720,7 @@ pub use async_graphql_derive::Union;
/// impl SubscriptionRoot {
/// async fn value(&self, condition: i32) -> impl Stream<Item = i32> {
/// // Returns the number from 0 to `condition`.
/// futures::stream::iter(0..condition)
/// futures_util::stream::iter(0..condition)
/// }
/// }
/// ```
@ -789,7 +789,7 @@ pub use async_graphql_derive::MergedObject;
///
/// ```rust
/// use async_graphql::*;
/// use futures::Stream;
/// use futures_util::stream::Stream;
///
/// #[derive(Default)]
/// struct Subscription1;
@ -797,7 +797,7 @@ pub use async_graphql_derive::MergedObject;
/// #[Subscription]
/// impl Subscription1 {
/// async fn events1(&self) -> impl Stream<Item = i32> {
/// futures::stream::iter(0..10)
/// futures_util::stream::iter(0..10)
/// }
/// }
///
@ -807,7 +807,7 @@ pub use async_graphql_derive::MergedObject;
/// #[Subscription]
/// impl Subscription2 {
/// async fn events2(&self) -> impl Stream<Item = i32> {
/// futures::stream::iter(10..20)
/// futures_util::stream::iter(10..20)
/// }
/// }
///

2
src/resolver_utils/container.rs
View File

@ -82,7 +82,7 @@ async fn resolve_container_inner<'a, T: ContainerType + Send + Sync>(
fields.add_set(ctx, root)?;
let res = if parallel {
futures::future::try_join_all(fields.0).await?
futures_util::future::try_join_all(fields.0).await?
} else {
let mut results = Vec::with_capacity(fields.0.len());
for field in fields.0 {

2
src/resolver_utils/list.rs
View File

@ -54,5 +54,5 @@ pub async fn resolve_list<'a, T: OutputValueType + Send + Sync + 'a>(
});
}
Ok(Value::List(futures::future::try_join_all(futures).await?))
Ok(Value::List(futures_util::future::try_join_all(futures).await?))
}

4
src/schema.rs
View File

@ -4,7 +4,7 @@ use std::ops::Deref;
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use futures::stream::{self, Stream, StreamExt};
use futures_util::stream::{self, Stream, StreamExt};
use indexmap::map::IndexMap;
use crate::context::{Data, QueryEnvInner, ResolveId};
@ -487,7 +487,7 @@ where
match batch_request {
BatchRequest::Single(request) => BatchResponse::Single(self.execute(request).await),
BatchRequest::Batch(requests) => BatchResponse::Batch(
futures::stream::iter(requests.into_iter())
futures_util::stream::iter(requests.into_iter())
.then(|request| self.execute(request))
.collect()
.await,

2
src/subscription.rs
View File

@ -1,6 +1,6 @@
use std::pin::Pin;
use futures::{Stream, StreamExt};
use futures_util::stream::{Stream, StreamExt};
use crate::parser::types::{Selection, TypeCondition};
use crate::{

2
src/types/connection/connection_type.rs
View File

@ -1,6 +1,6 @@
use std::borrow::Cow;
use futures::{Stream, StreamExt, TryStreamExt};
use futures_util::stream::{Stream, StreamExt, TryStreamExt};
use indexmap::map::IndexMap;
use crate::connection::edge::Edge;

2
src/types/empty_subscription.rs
View File

@ -1,7 +1,7 @@
use std::borrow::Cow;
use std::pin::Pin;
use futures::{stream, Stream};
use futures_util::stream::{self, Stream};
use crate::{registry, Context, ServerError, ServerResult, SubscriptionType, Type, Value};

2
src/types/upload.rs
View File

@ -2,7 +2,7 @@ use std::borrow::Cow;
use std::fs::File;
use std::io::Read;
use futures::AsyncRead;
use futures_util::io::AsyncRead;
use crate::{registry, Context, InputValueError, InputValueResult, InputValueType, Type, Value};

2
src/validation/suggestion.rs
View File

@ -46,7 +46,7 @@ where
for (i, s) in selected.iter().enumerate() {
if i != 0 {
suggestion.push_str(" , ");
suggestion.push_str(", ");
}
write!(suggestion, "\"{}\"", s).unwrap();
}

10
tests/field_features.rs
View File

@ -1,7 +1,7 @@
#![allow(dead_code)]
use async_graphql::*;
use futures::{Stream, StreamExt};
use futures_util::stream::{Stream, StreamExt};
#[async_std::test]
pub async fn test_field_features() {
@ -19,19 +19,19 @@ pub async fn test_field_features() {
#[Subscription]
impl SubscriptionRoot {
async fn values(&self) -> impl Stream<Item = i32> {
futures::stream::once(async move { 10 })
futures_util::stream::once(async move { 10 })
}
#[cfg(feature = "bson")]
async fn values_bson(&self) -> impl Stream<Item = i32> {
futures::stream::once(async move { 10 })
futures_util::stream::once(async move { 10 })
}
#[cfg(feature = "abc")]
async fn values_abc(
&self,
) -> Pin<Box<dyn async_graphql::futures::Stream<Item = i32> + Send + 'static>> {
Box::pin(futures::stream::once(async move { 10 }))
) -> Pin<Box<dyn Stream<Item = i32> + Send + 'static>> {
Box::pin(futures_util::stream::once(async move { 10 }))
}
}

4
tests/guard.rs
View File

@ -1,6 +1,6 @@
use async_graphql::guard::Guard;
use async_graphql::*;
use futures::{Stream, StreamExt};
use futures_util::stream::{Stream, StreamExt};
#[derive(Eq, PartialEq, Copy, Clone)]
enum Role {
@ -71,7 +71,7 @@ pub async fn test_guard_simple_rule() {
impl Subscription {
#[graphql(guard(RoleGuard(role = "Role::Admin")))]
async fn values(&self) -> impl Stream<Item = i32> {
futures::stream::iter(vec![1, 2, 3])
futures_util::stream::iter(vec![1, 2, 3])
}
}

6
tests/merged_object.rs
View File

@ -1,5 +1,5 @@
use async_graphql::*;
use futures::{Stream, StreamExt};
use futures_util::stream::{Stream, StreamExt};
#[derive(SimpleObject)]
struct Object1 {
@ -165,7 +165,7 @@ pub async fn test_merged_subscription() {
#[Subscription]
impl Subscription1 {
async fn events1(&self) -> impl Stream<Item = i32> {
futures::stream::iter(0..10)
futures_util::stream::iter(0..10)
}
}
@ -175,7 +175,7 @@ pub async fn test_merged_subscription() {
#[Subscription]
impl Subscription2 {
async fn events2(&self) -> impl Stream<Item = i32> {
futures::stream::iter(10..20)
futures_util::stream::iter(10..20)
}
}

2
tests/mutation.rs
View File

@ -1,5 +1,5 @@
use async_graphql::*;
use futures::lock::Mutex;
use async_std::sync::Mutex;
use std::sync::Arc;
use std::time::Duration;

4
tests/raw_ident.rs
View File

@ -1,5 +1,5 @@
use async_graphql::*;
use futures::{Stream, StreamExt, TryStreamExt};
use futures_util::stream::{Stream, StreamExt, TryStreamExt};
#[async_std::test]
pub async fn test_input_value_custom_error() {
@ -41,7 +41,7 @@ pub async fn test_input_value_custom_error() {
#[Subscription]
impl SubscriptionRoot {
async fn r#type(&self) -> impl Stream<Item = i32> {
futures::stream::iter(0..10)
futures_util::stream::iter(0..10)
}
}

4
tests/rename.rs
View File

@ -1,5 +1,5 @@
use async_graphql::*;
use futures::{Stream, StreamExt};
use futures_util::stream::{Stream, StreamExt};
#[async_std::test]
pub async fn test_enum() {
@ -114,7 +114,7 @@ pub async fn test_subscription() {
#[allow(non_snake_case)]
impl Subscription {
async fn create_object(&self, ObjectId: i32) -> impl Stream<Item = i32> {
futures::stream::once(async move { ObjectId })
futures_util::stream::once(async move { ObjectId })
}
}

24
tests/subscription.rs
View File

@ -1,5 +1,5 @@
use async_graphql::*;
use futures::{Stream, StreamExt, TryStreamExt};
use futures_util::stream::{Stream, StreamExt, TryStreamExt};
#[async_std::test]
pub async fn test_subscription() {
@ -19,11 +19,11 @@ pub async fn test_subscription() {
#[Subscription]
impl SubscriptionRoot {
async fn values(&self, start: i32, end: i32) -> impl Stream<Item = i32> {
futures::stream::iter(start..end)
futures_util::stream::iter(start..end)
}
async fn events(&self, start: i32, end: i32) -> impl Stream<Item = Event> {
futures::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
futures_util::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
}
}
@ -77,11 +77,11 @@ pub async fn test_subscription_with_ctx_data() {
impl SubscriptionRoot {
async fn values(&self, ctx: &Context<'_>) -> impl Stream<Item = i32> {
let value = *ctx.data_unchecked::<i32>();
futures::stream::once(async move { value })
futures_util::stream::once(async move { value })
}
async fn objects(&self) -> impl Stream<Item = MyObject> {
futures::stream::once(async move { MyObject })
futures_util::stream::once(async move { MyObject })
}
}
@ -118,7 +118,7 @@ pub async fn test_subscription_with_token() {
if ctx.data_unchecked::<Token>().0 != "123456" {
return Err("forbidden".into());
}
Ok(futures::stream::once(async move { 100 }))
Ok(futures_util::stream::once(async move { 100 }))
}
}
@ -166,7 +166,7 @@ pub async fn test_subscription_inline_fragment() {
#[Subscription]
impl SubscriptionRoot {
async fn events(&self, start: i32, end: i32) -> impl Stream<Item = Event> {
futures::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
futures_util::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
}
}
@ -219,7 +219,7 @@ pub async fn test_subscription_fragment() {
#[Subscription]
impl SubscriptionRoot {
async fn events(&self, start: i32, end: i32) -> impl Stream<Item = Event> {
futures::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
futures_util::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
}
}
@ -274,7 +274,7 @@ pub async fn test_subscription_fragment2() {
#[Subscription]
impl SubscriptionRoot {
async fn events(&self, start: i32, end: i32) -> impl Stream<Item = Event> {
futures::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
futures_util::stream::iter((start..end).map(|n| Event { a: n, b: n * 10 }))
}
}
@ -333,7 +333,7 @@ pub async fn test_subscription_error() {
#[Subscription]
impl SubscriptionRoot {
async fn events(&self) -> impl Stream<Item = Event> {
futures::stream::iter((0..10).map(|n| Event { value: n }))
futures_util::stream::iter((0..10).map(|n| Event { value: n }))
}
}
@ -380,9 +380,9 @@ pub async fn test_subscription_fieldresult() {
#[Subscription]
impl SubscriptionRoot {
async fn values(&self) -> impl Stream<Item = Result<i32>> {
futures::stream::iter(0..5)
futures_util::stream::iter(0..5)
.map(Result::Ok)
.chain(futures::stream::once(
.chain(futures_util::stream::once(
async move { Err("StreamErr".into()) },
))
}

17
tests/subscription_websocket.rs
View File

@ -1,6 +1,5 @@
use async_graphql::*;
use futures::channel::mpsc;
use futures::{SinkExt, Stream, StreamExt};
use futures_util::stream::{Stream, StreamExt};
#[async_std::test]
pub async fn test_subscription_ws_transport() {
@ -14,12 +13,12 @@ pub async fn test_subscription_ws_transport() {
#[Subscription]
impl SubscriptionRoot {
async fn values(&self) -> impl Stream<Item = i32> {
futures::stream::iter(0..10)
futures_util::stream::iter(0..10)
}
}
let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
let (mut tx, rx) = mpsc::unbounded();
let (tx, rx) = async_channel::unbounded();
let mut stream = http::WebSocket::new(schema, rx);
tx.send(
@ -88,12 +87,12 @@ pub async fn test_subscription_ws_transport_with_token() {
if ctx.data_unchecked::<Token>().0 != "123456" {
return Err("forbidden".into());
}
Ok(futures::stream::iter(0..10))
Ok(futures_util::stream::iter(0..10))
}
}
let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
let (mut tx, rx) = mpsc::unbounded();
let (tx, rx) = async_channel::unbounded();
let mut stream = http::WebSocket::with_data(
schema,
rx,
@ -187,12 +186,12 @@ pub async fn test_subscription_ws_transport_error() {
#[Subscription]
impl SubscriptionRoot {
async fn events(&self) -> impl Stream<Item = Event> {
futures::stream::iter((0..10).map(|n| Event { value: n }))
futures_util::stream::iter((0..10).map(|n| Event { value: n }))
}
}
let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
let (mut tx, rx) = mpsc::unbounded();
let (tx, rx) = async_channel::unbounded();
let mut stream = http::WebSocket::new(schema, rx);
tx.send(
@ -264,7 +263,7 @@ pub async fn test_query_over_websocket() {
}
let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
let (mut tx, rx) = mpsc::unbounded();
let (tx, rx) = async_channel::unbounded();
let mut stream = http::WebSocket::new(schema, rx);
tx.send(