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

use crate::{Stream, StreamExt, sender::SenderInner, utils::transform_sink_lifetime};
use std::rc::Rc;
use std::cell::UnsafeCell;
use priv_traits::Remove;


pub struct StreamVec<'a,T>{
    inner: Rc<UnsafeCell<SenderInner<'a, T>>>,
    removes: Rc<UnsafeCell<Vec<Rc<UnsafeCell<dyn Remove>>>>>
}

impl<'a, T: 'a> StreamVec<'a, T> {
    pub fn new() -> (StreamVec<'a, T>, Stream<'a, T>) {
        let inner = Rc::new(UnsafeCell::new(SenderInner::new()));
        let removes: Rc<UnsafeCell<Vec<Rc<UnsafeCell<dyn Remove>>>>> = Rc::new(UnsafeCell::new(vec![]));
        let stream = Stream::from_remove(removes.clone(), inner.clone());
        let stream_vec = StreamVec{
            inner,
            removes,
        };

        (stream_vec, stream)
    }

    pub fn push_source<S: StreamExt<'a, T>>(&mut self, stream: S) {
        unsafe{
            (*stream.inner_rc().get()).push(transform_sink_lifetime(self.inner.clone()));
            (*self.removes.get()).push( std::mem::transmute(stream.into_remove()) );
        }
    }
}