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 ([email protected])",
    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 {
      // ...
  }
}

Answer 1

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| ...)