CODEWITHSUNDEEP
rustrust-futures
eltiare
eltiare

Reputation: 1939

How can I wrap a dynamically typed stream for API convenience?

I'm looking to implement a wrapper struct for any stream that returns a certain type, to cut down on the dynamic keywords littering my application. I've come across BoxStream, but have no idea how to make use of it in Stream::poll_next. Here's what I have so far:

use std::pin::Pin;
use std::task::{Context, Poll};
use futures::prelude::stream::BoxStream;
use futures::Stream;

pub struct Row;

pub struct RowCollection<'a> {
    stream: BoxStream<'a, Row>,
}

impl RowCollection<'_> {
    pub fn new<'a>(stream: BoxStream<Row>) -> RowCollection {
        RowCollection { stream }
    }
}

impl Stream for RowCollection<'_> {
    type Item = Row;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // I have no idea what to put here, but it needs to get the information from self.stream and return appropriate value
    }
}

Dependencies:

futures = "0.3"

Upvotes: 2

Views: 565

Answers (1)

Shepmaster
Shepmaster

Reputation: 432109

Since Box implements Unpin, then BoxStream implements Unpin, and so will RowCollection.

Because of this, you can make use of Pin::get_mut which will give you a &mut RowCollection. From that, you can get a &mut BoxStream. You can re-pin that via Pin::new and then call poll_next on it. This is called pin-projection.

impl Stream for RowCollection<'_> {
    type Item = Row;
    
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut self.get_mut().stream).poll_next(cx)
    }
}

See also:

Upvotes: 5

Related Questions