rust - 暗示AsyncRead为补品::流

Question

我正在尝试服用补品routeguide tutorial，并将客户端变成rocket服务器。我只是接受响应并将gRPC转换为字符串。

service RouteGuide {
    rpc GetFeature(Point) returns (Feature) {}
    rpc ListFeatures(Rectangle) returns (stream Feature) {}
}

这对于GetFeature来说已经足够好了。对于ListFeatures查询，就像Tonic允许客户端响应中的流一样，我想将此传递给Rocket客户端。我看到Rocket支持 streaming响应，但是我需要实现 AsyncRead特性。
有什么办法可以做这样的事情吗？以下是关于我在做什么的精简版:

struct FeatureStream {
    stream: tonic::Streaming<Feature>,
}

impl AsyncRead for FeatureStream {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        // Write out as utf8 any response messages.
        match Pin::new(&mut self.stream.message()).poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(feature) => Poll::Pending,
        }
    }
}

#[get("/list_features")]
async fn list_features(client: State<'_, RouteGuideClient<Channel>>) -> Stream<FeatureStream> {
    let rectangle = Rectangle {
        low: Some(Point {
            latitude: 400_000_000,
            longitude: -750_000_000,
        }),
        high: Some(Point {
            latitude: 420_000_000,
            longitude: -730_000_000,
        }),
    };
    let mut client = client.inner().clone();
    let stream = client
        .list_features(Request::new(rectangle))
        .await
        .unwrap()
        .into_inner();
    Stream::from(FeatureStream { stream })
}

#[rocket::launch]
async fn rocket() -> rocket::Rocket {
    rocket::ignite()
        .manage(
            create_route_guide_client("http://[::1]:10000")
                .await
                .unwrap(),
        )
        .mount("/", rocket::routes![list_features,])
}

与错误:

error[E0277]: `from_generator::GenFuture<[static generator@Streaming<Feature>::message::{closure#0} for<'r, 's, 't0, 't1, 't2> {ResumeTy, &'r mut Streaming<Feature>, [closure@Streaming<Feature>::message::{closure#0}::{closure#0}], rocket::futures::future::PollFn<[closure@Streaming<Feature>::message::{closure#0}::{closure#0}]>, ()}]>` cannot be unpinned
   --> src/web_user.rs:34:15
    |
34  |         match Pin::new(&mut self.stream.message()).poll(cx) {
    |               ^^^^^^^^ within `impl std::future::Future`, the trait `Unpin` is not implemented for `from_generator::GenFuture<[static generator@Streaming<Feature>::message::{closure#0} for<'r, 's, 't0, 't1, 't2> {ResumeTy, &'r mut Streaming<Feature>, [closure@Streaming<Feature>::message::{closure#0}::{closure#0}], rocket::futures::future::PollFn<[closure@Streaming<Feature>::message::{closure#0}::{closure#0}]>, ()}]>`
    | 
   ::: /home/matan/.cargo/registry/src/github.com-1ecc6299db9ec823/tonic-0.4.0/src/codec/decode.rs:106:40
    |
106 |     pub async fn message(&mut self) -> Result<Option<T>, Status> {
    |                                        ------------------------- within this `impl std::future::Future`
    |
    = note: required because it appears within the type `impl std::future::Future`
    = note: required because it appears within the type `impl std::future::Future`
    = note: required by `Pin::<P>::new`

Answer 1

问题是从Future生成的tonic::Streaming<Feature>::message()没有实现Unpin，因为它是异​​步函数。让我们将此类型标记为MessageFuture，因为取消引用的类型&mut MessageFuture不实现MessageFuture，所以您无法安全地将Unpin指针固定为。
为什么不安全？
从reference，Unpin的实现带来了:

Types that can be safely moved after being pinned.

这意味着如果 T:!Unpin然后 Pin<&mut T>无法移动，这很重要，因为由异步块创建的 Future没有 Unpin实现，因为它可能持有自身的成员引用，并且如果您移动 T，则该引用的指针也将被移动。 ，但引用仍指向相同的地址，以防止该地址不能移动。请阅读 "Pinning" section from async-book以查看原因。
注意: T:!Unpin表示 T是没有 Unpin实现的类型。
解决方案 message()函数可帮助您从 tonic::Streaming<T>中选择下一条消息。您不需要特别调用 message()从流中选择下一个元素，因为结构中已经有实际的流。

struct FeatureStream {stream: tonic::Streaming<Feature>}

您可以等待 AsyncRead的下一条消息，例如:

impl AsyncRead for FeatureStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {

       //it returns Pending for all cases as your code does, you can change it as you want
        match self.stream.poll_next_unpin(cx) {
            Poll::Ready(Some(Ok(m))) => Poll::Pending,
            Poll::Ready(Some(Err(e))) => Poll::Pending,
            Poll::Ready(None) => Poll::Pending,
            Poll::Pending => Poll::Pending
        }
    }
}

请注意， tonic::Streaming<T>具有 Unpin( reference)的实现。