impl Stream cannot be unpinned

Question

I'm trying to get data using crates_io_api. I attempted to get data from a stream, but I can not get it to work.

AsyncClient::all_crates returns an impl Stream. How do I get data from it? It would be helpful if you provide code.

I checked out the async book but it didn't work. Thank you.

Here's my current code.

use crates_io_api::{AsyncClient, Error};
use futures::stream::StreamExt;

async fn get_all(query: Option<String>) -> Result<crates_io_api::Crate, Error> {
  // Instantiate the client.
  let client = AsyncClient::new(
    "test (test@test.com)",
    std::time::Duration::from_millis(10000),
  )?;

  let stream = client.all_crates(query);

  // what should I do after?
  // ERROR: `impl Stream cannot be unpinned`
  while let Some(item) = stream.next().await {
      // ...
  }
}

https://docs.rs/futures/0.3.14/futures/stream/trait.StreamExt.html#method.next — Ibraheem Ahmed, May 07 '21 at 13:57
@IbraheemAhmed Thanks. I tried it but got long errors. I updated the code in the post — Pytan, May 07 '21 at 14:02
This is likely a mistake on the side of `crates_io_api`. They should be returning an `impl Stream + Unpin`. — Ibraheem Ahmed, May 07 '21 at 14:11

Ibraheem Ahmed · Accepted Answer · 2021-05-07T17:31:25.630

This looks like a mistake on the side of crates_io_api. Getting the next element of a Stream requires that the Stream is Unpin:

pub fn next(&mut self) -> Next<'_, Self> where
    Self: Unpin,

Because Next stores a reference to Self, you must guarantee that Self is not moved during the process, or risk pointer invalidation. This is what the Unpin marker trait represents. crates_io_api does not provide this guarantee (although they could, and should be), so you must make it yourself. To convert a !Unpin type to a Unpin type, you can pin it to a heap allocation:

use futures::stream::StreamExt;

let stream = client.all_crates(query).boxed();

// boxed simply calls Box::pin
while let Some(elem) = stream.next() { ... }

Or you can pin it to the stack with the pin_mut!/pin! macro:

let stream = client.all_crates(query);
futures::pin_mut!(stream);

while let Some(elem) = stream.next() { ... }

Alternatively, you could use a combinator that does not require Unpin such as for_each:

stream.for_each(|elem| ...)

Thank you! That works. I didnt know about Pin, so this is a good time to study — Pytan, May 07 '21 at 14:30
There's a great blog post explaining the concept of pinning and why it's necessary: https://fasterthanli.me/articles/pin-and-suffering — sebpuetz, May 08 '21 at 00:16

impl Stream cannot be unpinned

1 Answers1

Linked