Skip to content

Commit 6c4c958

Browse files
committed
from/into stream
Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> update examples Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> impl collect Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> compiles! Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> layout base for collect into vec Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> fmt Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> progress Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> compiles! Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> define failing test Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> cargo fmt Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> stuck again Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> fix trait bounds! Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> cargo fmt Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> hide dyn fut impl Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> dyn ret for vec Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> cargo fmt Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> collect docs Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> remove macro from vec::from_stream Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> shorten collect trait bound Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com> Remove some Unpin and Send bounds Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com>
1 parent 60a62f9 commit 6c4c958

File tree

7 files changed

+162
-3
lines changed

7 files changed

+162
-3
lines changed

src/lib.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ pub mod prelude;
5151
pub mod stream;
5252
pub mod sync;
5353
pub mod task;
54+
mod vec;
5455

5556
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
5657
#[cfg(feature = "unstable")]

src/stream/from_stream.rs

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
use super::IntoStream;
2+
3+
use std::pin::Pin;
4+
5+
/// Conversion from a `Stream`.
6+
///
7+
/// By implementing `FromStream` for a type, you define how it will be created from a stream.
8+
/// This is common for types which describe a collection of some kind.
9+
///
10+
/// See also: [`IntoStream`].
11+
///
12+
/// [`IntoStream`]: trait.IntoStream.html
13+
pub trait FromStream<T> {
14+
/// Creates a value from a stream.
15+
///
16+
/// # Examples
17+
///
18+
/// Basic usage:
19+
///
20+
/// ```
21+
/// // use async_std::stream::FromStream;
22+
///
23+
/// // let _five_fives = async_std::stream::repeat(5).take(5);
24+
/// ```
25+
fn from_stream<'a, S: IntoStream<Item = T> + 'a>(
26+
stream: S,
27+
) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>;
28+
}

src/stream/into_stream.rs

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
use futures_core::stream::Stream;
2+
3+
/// Conversion into a `Stream`.
4+
///
5+
/// By implementing `IntoIterator` for a type, you define how it will be
6+
/// converted to an iterator. This is common for types which describe a
7+
/// collection of some kind.
8+
///
9+
/// [`from_stream`]: #tymethod.from_stream
10+
/// [`Stream`]: trait.Stream.html
11+
/// [`collect`]: trait.Stream.html#method.collect
12+
///
13+
/// See also: [`FromStream`].
14+
///
15+
/// [`FromStream`]: trait.FromStream.html
16+
pub trait IntoStream {
17+
/// The type of the elements being iterated over.
18+
type Item;
19+
20+
/// Which kind of stream are we turning this into?
21+
type IntoStream: Stream<Item = Self::Item>;
22+
23+
/// Creates a stream from a value.
24+
fn into_stream(self) -> Self::IntoStream;
25+
}
26+
27+
impl<I: Stream> IntoStream for I {
28+
type Item = I::Item;
29+
type IntoStream = I;
30+
31+
#[inline]
32+
fn into_stream(self) -> I {
33+
self
34+
}
35+
}

src/stream/mod.rs

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,12 +23,16 @@
2323
2424
pub use double_ended_stream::DoubleEndedStream;
2525
pub use empty::{empty, Empty};
26+
pub use from_stream::FromStream;
27+
pub use into_stream::IntoStream;
2628
pub use once::{once, Once};
2729
pub use repeat::{repeat, Repeat};
2830
pub use stream::{Scan, Stream, Take, Zip};
2931

3032
mod double_ended_stream;
3133
mod empty;
34+
mod from_stream;
35+
mod into_stream;
3236
mod once;
3337
mod repeat;
3438
mod stream;

src/stream/stream/mod.rs

Lines changed: 60 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,14 +50,15 @@ use min_by::MinByFuture;
5050
use next::NextFuture;
5151
use nth::NthFuture;
5252

53+
use super::from_stream::FromStream;
54+
use crate::future::Future;
55+
use crate::task::{Context, Poll};
5356
use std::cmp::Ordering;
5457
use std::marker::PhantomData;
5558
use std::pin::Pin;
5659

5760
use cfg_if::cfg_if;
5861

59-
use crate::task::{Context, Poll};
60-
6162
cfg_if! {
6263
if #[cfg(feature = "docs")] {
6364
#[doc(hidden)]
@@ -80,6 +81,21 @@ cfg_if! {
8081
}
8182
}
8283

84+
cfg_if! {
85+
if #[cfg(feature = "docs")] {
86+
#[doc(hidden)]
87+
pub struct DynFuture<'a, T>(std::marker::PhantomData<&'a T>);
88+
89+
macro_rules! dyn_ret {
90+
($a:lifetime, $o:ty) => (DynFuture<$a, $o>);
91+
}
92+
} else {
93+
macro_rules! dyn_ret {
94+
($a:lifetime, $o:ty) => (Pin<Box<dyn core::future::Future<Output = $o> + 'a>>)
95+
}
96+
}
97+
}
98+
8399
/// An asynchronous stream of values.
84100
///
85101
/// This trait is an async version of [`std::iter::Iterator`].
@@ -528,7 +544,6 @@ pub trait Stream {
528544
///
529545
/// Basic usage:
530546
///
531-
/// ```
532547
/// # fn main() { async_std::task::block_on(async {
533548
/// #
534549
/// use async_std::prelude::*;
@@ -652,6 +667,48 @@ pub trait Stream {
652667
{
653668
Zip::new(self, other)
654669
}
670+
671+
/// Transforms a stream into a collection.
672+
///
673+
/// `collect()` can take anything streamable, and turn it into a relevant
674+
/// collection. This is one of the more powerful methods in the async
675+
/// standard library, used in a variety of contexts.
676+
///
677+
/// The most basic pattern in which `collect()` is used is to turn one
678+
/// collection into another. You take a collection, call [`stream`] on it,
679+
/// do a bunch of transformations, and then `collect()` at the end.
680+
///
681+
/// Because `collect()` is so general, it can cause problems with type
682+
/// inference. As such, `collect()` is one of the few times you'll see
683+
/// the syntax affectionately known as the 'turbofish': `::<>`. This
684+
/// helps the inference algorithm understand specifically which collection
685+
/// you're trying to collect into.
686+
///
687+
/// # Examples
688+
///
689+
/// ```
690+
/// # fn main() { async_std::task::block_on(async {
691+
/// #
692+
/// use async_std::prelude::*;
693+
/// use async_std::stream;
694+
///
695+
/// let s = stream::repeat(9u8).take(3);
696+
/// let buf: Vec<u8> = s.collect().await;
697+
///
698+
/// assert_eq!(buf, vec![9; 3]);
699+
/// #
700+
/// # }) }
701+
/// ```
702+
///
703+
/// [`stream`]: trait.Stream.html#tymethod.next
704+
#[must_use = "if you really need to exhaust the iterator, consider `.for_each(drop)` instead (TODO)"]
705+
fn collect<'a, B>(self) -> dyn_ret!('a, B)
706+
where
707+
Self: futures_core::stream::Stream + Sized + 'a,
708+
B: FromStream<<Self as futures_core::stream::Stream>::Item>,
709+
{
710+
FromStream::from_stream(self)
711+
}
655712
}
656713

657714
impl<T: futures_core::stream::Stream + ?Sized> Stream for T {

src/vec/from_stream.rs

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
use crate::stream::{FromStream, IntoStream, Stream};
2+
3+
use std::pin::Pin;
4+
5+
impl<T> FromStream<T> for Vec<T> {
6+
#[inline]
7+
fn from_stream<'a, S: IntoStream<Item = T>>(
8+
stream: S,
9+
) -> Pin<Box<dyn core::future::Future<Output = Self> + 'a>>
10+
where
11+
<S as IntoStream>::IntoStream: 'a,
12+
{
13+
let stream = stream.into_stream();
14+
15+
Pin::from(Box::new(async move {
16+
pin_utils::pin_mut!(stream);
17+
18+
let mut out = vec![];
19+
while let Some(item) = stream.next().await {
20+
out.push(item);
21+
}
22+
out
23+
}))
24+
}
25+
}

src/vec/mod.rs

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
//! The Rust core allocation and collections library
2+
//!
3+
//! This library provides smart pointers and collections for managing
4+
//! heap-allocated values.
5+
6+
mod from_stream;
7+
8+
#[doc(inline)]
9+
pub use std::vec::Vec;

0 commit comments

Comments
 (0)