Added skip and Skip struct to parallel stream. #12

justin1dennison · 2020-03-23T21:20:40Z

No description provided.

Thegaram · 2020-03-24T08:07:00Z

src/par_stream/skip.rs

+#[async_std::test]
+async fn smoke() {
+    use async_std::prelude::*;
+    let s = async_std::stream::from_iter(vec![1, 2, 3, 4, 5, 6]).skip(3);


Your test is calling async_std::stream::Stream, not ParallelStream::skip.

Thegaram · 2020-03-24T08:07:34Z

src/par_stream/skip.rs

+    /// This `struct` is created by the [`skip`] method on [`ParallelStream`]. See its
+    /// documentation for more.
+    ///
+    /// [`skip`]: trait.ParallelStream.html#method.take


I think this should be trait.ParallelStream.html#method.skip.

justin1dennison · 2020-03-24T13:44:37Z

I don't know if I am way off base with this implementation. I am having the following issue:


--> src/par_stream/skip.rs:42:23
   |
42 |         Skip { limit, receiver }
   |                       ^^^^^^^^ expected type parameter `T`, found associated type
   |
   = note: expected struct `async_std::sync::channel::Receiver<T>`
              found struct `async_std::sync::channel::Receiver<<S as par_stream::ParallelStream>::Item>`
   = note: you might be missing a type parameter or trait bound

Thegaram · 2020-03-24T13:53:22Z

src/par_stream/skip.rs

+impl<T: Send + 'static> Skip<T> {
+    pub(super) fn new<S>(mut stream: S, mut skipped: usize) -> Self
+    where
+        S: ParallelStream
+    {
+        let limit = stream.get_limit();
+        let (sender, receiver) = sync::channel(1);
+        task::spawn(async move {
+            while let Some(val) = stream.next().await {
+                if skipped == 0 {
+                    sender.send(val).await
+                } else {
+                    skipped -= 1;
+                }
+            }
+        });
+
+        Skip { limit, receiver }
+    }
+}
+
+impl<T: Send + 'static> ParallelStream for Skip<T> {
+    type Item = T;


You could use the same approach as Take:

impl<S: ParallelStream> Skip<S> { ... } impl<S: ParallelStream> ParallelStream for Skip<S> { type Item = S::Item; ... }

Otherwise, the compiler will think you're trying to transform a ParallelStream<Item = X> into a ParallelStream<Item = T> and it will not be able to figure out what T is.

completed implementation of skip

3df85e1

Thegaram reviewed Mar 24, 2020

View reviewed changes

fixed some errors with typos and misuse of when testing

921fed3

Thegaram reviewed Mar 24, 2020

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added skip and Skip struct to parallel stream. #12

Added skip and Skip struct to parallel stream. #12

justin1dennison commented Mar 23, 2020

Thegaram Mar 24, 2020

Thegaram Mar 24, 2020

justin1dennison commented Mar 24, 2020 •

edited

Loading

Thegaram Mar 24, 2020

Added skip and Skip struct to parallel stream. #12

Are you sure you want to change the base?

Added skip and Skip struct to parallel stream. #12

Conversation

justin1dennison commented Mar 23, 2020

Thegaram Mar 24, 2020

Choose a reason for hiding this comment

Thegaram Mar 24, 2020

Choose a reason for hiding this comment

justin1dennison commented Mar 24, 2020 • edited Loading

Thegaram Mar 24, 2020

Choose a reason for hiding this comment

justin1dennison commented Mar 24, 2020 •

edited

Loading