deluge 0.2.1

A highly concurrent stream library driving the underlying futures either concurrently or in parallel to process streaming operations as quickly as possible.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

use crate::deluge::Deluge;

use std::cell::RefCell;
use std::collections::VecDeque;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;

pub struct PreloadedFutures<'a, Del>
where
    Del: Deluge + 'a,
{
    storage: RefCell<VecDeque<Pin<Box<Del::Output<'a>>>>>,
    _del: PhantomData<Del>,
}

impl<'a, Del> PreloadedFutures<'a, Del>
where
    Del: Deluge + 'a,
{
    pub fn new(deluge: Del) -> Self {
        let mut storage = VecDeque::new();
        let deluge_borrow: &'a Del = unsafe { std::mem::transmute(&deluge) };
        while let Some(v) = deluge_borrow.next() {
            storage.push_back(Box::pin(v));
        }

        Self {
            storage: RefCell::new(storage),
            _del: PhantomData,
        }
    }

    pub fn len(&self) -> usize {
        self.storage.borrow().len()
    }
}

impl<'a, Del> Deluge for PreloadedFutures<'a, Del>
where
    Del: Deluge + 'a,
{
    type Item = Del::Item;
    type Output<'x> = impl Future<Output = Option<Self::Item>> + 'x where Self: 'x;

    fn next(&self) -> Option<Self::Output<'_>> {
        let next_item = { self.storage.borrow_mut().pop_front() };

        next_item.map(|el| async move { el.await })
    }
}