如何限制 futures::Stream::concat2 读取的字节数？

Question

对的回答表明：

you may wish to establish some kind of cap on the number of bytes read [when using futures::Stream::concat2]

我怎样才能真正做到这一点？例如，下面是一些模仿向我的服务发送无限量数据的恶意用户的代码：

extern crate futures; // 0.1.25

use futures::{prelude::*, stream};

fn some_bytes() -> impl Stream<Item = Vec<u8>, Error = ()> {
    stream::repeat(b"0123456789ABCDEF".to_vec())
}

fn limited() -> impl Future<Item = Vec<u8>, Error = ()> {
    some_bytes().concat2()
}

fn main() {
    let v = limited().wait().unwrap();
    println!("{}", v.len());
}

Answer 1

一个解决方案是创建一个流组合器，一旦超过某个字节阈值就结束流。这是一种可能的实施方式：

struct TakeBytes<S> {
    inner: S,
    seen: usize,
    limit: usize,
}

impl<S> Stream for TakeBytes<S>
where
    S: Stream<Item = Vec<u8>>,
{
    type Item = Vec<u8>;
    type Error = S::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        if self.seen >= self.limit {
            return Ok(Async::Ready(None)); // Stream is over
        }

        let inner = self.inner.poll();
        if let Ok(Async::Ready(Some(ref v))) = inner {
            self.seen += v.len();
        }
        inner
    }
}

trait TakeBytesExt: Sized {
    fn take_bytes(self, limit: usize) -> TakeBytes<Self>;
}

impl<S> TakeBytesExt for S
where
    S: Stream<Item = Vec<u8>>,
{
    fn take_bytes(self, limit: usize) -> TakeBytes<Self> {
        TakeBytes {
            inner: self,
            limit,
            seen: 0,
        }
    }
}

然后可以在 concat2:

之前将其链接到流中

fn limited() -> impl Future<Item = Vec<u8>, Error = ()> {
    some_bytes().take_bytes(999).concat2()
}

此实现有一些注意事项：

它仅适用于 Vec<u8>。当然，你可以引入泛型使其更广泛地适用。
它允许超过限制的字节数进入，它只是在该点之后停止流。这些类型的决定取决于应用程序。

另一件要记住的事情是，你想尝试尽可能低地解决这个问题——如果数据源已经分配了 1 GB 的内存，那么设置限制将无济于事.

如何限制 futures::Stream::concat2 读取的字节数？

How do I apply a limit to the number of bytes read by futures::Stream::concat2?

future

rust