impl Stream 不能取消固定

Question

我正在尝试使用crates_io_api. 我试图从流中获取数据，但我无法让它工作。

AsyncClient::all_crates返回一个impl Stream. 我如何从中获取数据？如果你提供代码会很有帮助。

我检查了异步书，但没有用。谢谢你。

这是我当前的代码。

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 {
      // ...
  }
}

Answer 1

这看起来像是 方面的一个错误crates_io_api。获取nexta 的元素Stream需要Streamis Unpin：

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

因为Next存储了对 的引用Self，所以必须保证Self在此过程中不被移动，否则存在指针失效的风险。这就是Unpin标记特征所代表的内容。crates_io_api不提供此保证（尽管他们可以而且应该提供），因此您必须自己做出保证。要将!Unpin类型转换为Unpin类型，您可以将其固定到堆分配：

use futures::stream::StreamExt;

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

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

pin_mut!或者您可以使用/宏将其固定到堆栈pin!：

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

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

或者，您可以使用不需要的组合器，Unpin例如for_each：

stream.for_each(|elem| ...)