oxydsp/oxydsp-flowgraph/src/edge.rs

use std::any::Any;
use std::collections::binary_heap::Iter;
use std::sync::Arc;
use std::sync::Mutex;

use oxydsp_flowgraph_macros::generate_pop_iterable_tuple_impl;
use oxydsp_flowgraph_macros::impl_iterator_for_pop_iter_tuple;

use crate::stream;
use crate::stream::StreamConsumer;
use crate::stream::StreamProducer;
use crate::stream::StreamReader;
use crate::stream::StreamWriter;

#[derive(Default)]
pub struct Edge
{
    // Represents the index of the block owning the Out end in the graph
    // And the the output index within that block
    pub from: Option<BlockIOIndex>,

    // Represents the index of the block owning the In end in the graph
    // And the the input index within that block
    pub to: Option<BlockIOIndex>,
}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct BlockIOIndex
{
    pub block_index: usize,
    pub port_index: usize,
}

// Needed for graph to be able to manipulate
// stream endings without knowing the generic type
pub struct AnonymousStreamProducer
{
    inner: Box<dyn Any>,
}

pub struct AnonymousStreamConsumer
{
    inner: Box<dyn Any>,
}

impl<T: 'static> From<StreamProducer<T>> for AnonymousStreamProducer
{
    fn from(value: StreamProducer<T>) -> Self
    {
        AnonymousStreamProducer {
            inner: Box::new(value),
        }
    }
}

impl<T: 'static> From<StreamConsumer<T>> for AnonymousStreamConsumer
{
    fn from(value: StreamConsumer<T>) -> Self
    {
        AnonymousStreamConsumer {
            inner: Box::new(value),
        }
    }
}

impl AnonymousStreamProducer
{
    pub fn downcast<T: 'static>(self) -> StreamProducer<T>
    {
        *self.inner.downcast::<StreamProducer<T>>().unwrap()
    }
}

impl AnonymousStreamConsumer
{
    pub fn downcast<T: 'static>(self) -> StreamConsumer<T>
    {
        *self.inner.downcast::<StreamConsumer<T>>().unwrap()
    }
}

pub struct In<T>
{
    stream: Option<StreamConsumer<T>>,

    // Will rarely be accessed
    edge: Arc<Mutex<Edge>>,
}

pub struct Out<T>
{
    stream: Option<StreamProducer<T>>,

    // Will rarely be accessed
    edge: Arc<Mutex<Edge>>,
}

pub fn stream<T>() -> (Out<T>, In<T>)
{
    let edge = Arc::new(Mutex::new(Edge::default()));
    (
        Out {
            stream: None,
            edge: edge.clone(),
        },
        In { stream: None, edge },
    )
}

impl<T: 'static> In<T>
{
    pub fn set_block_index(&self, index: BlockIOIndex)
    {
        self.edge.lock().unwrap().to = Some(index);
    }

    pub fn get_producer_block(&self) -> Option<BlockIOIndex>
    {
        self.edge.lock().unwrap().from
    }

    pub fn set_anonymous_stream(&mut self, consumer: AnonymousStreamConsumer)
    {
        self.stream = Some(consumer.downcast::<T>())
    }

    pub fn read<'a>(&'a mut self) -> StreamReader<'a, T>
    {
        self.stream.as_mut().unwrap().read()
    }
}

impl<T: 'static> Out<T>
{
    pub fn set_block_index(&self, index: BlockIOIndex)
    {
        self.edge.lock().unwrap().from = Some(index);
    }

    pub fn get_consumer_block(&self) -> Option<BlockIOIndex>
    {
        self.edge.lock().unwrap().to
    }

    pub fn set_anonymous_stream(&mut self, producer: AnonymousStreamProducer)
    {
        self.stream = Some(producer.downcast::<T>())
    }

    // Delegate stream creation to Out object
    // which knows the stream type
    pub fn create_anonymous_stream(
        &self,
        capacity: usize,
    ) -> (AnonymousStreamProducer, AnonymousStreamConsumer)
    {
        let (tx, rx) = stream::bounded_queue::<T>(capacity);
        (tx.into(), rx.into())
    }

    pub fn write<'a>(&'a mut self) -> StreamWriter<'a, T>
    {
        self.stream.as_mut().unwrap().write()
    }

    pub fn push_iter<I: Iterator<Item = T>>(&mut self, mut iter: I) -> bool
    {
        let writer = self.write();
        let len = writer.len();

        for _ in 0..len
        {
            if let Some(elt) = iter.next()
            {
                let _ = writer.push(elt);
            }
            else
            {
                return false;
            }
        }
        true
    }

    // Meta information
    pub fn get_type_name(&self) -> &'static str
    {
        std::any::type_name::<T>()
    }
}

pub struct PopIter<T>
{
    len: usize,
    popped: usize,
    reader: T,
}

pub trait PopIterable<'a>
{
    type Output;
    fn pop_iter(&'a mut self) -> PopIter<Self::Output>;
}

impl<'a, T: 'static> PopIterable<'a> for In<T>
{
    type Output = StreamReader<'a, T>;
    fn pop_iter(&'a mut self) -> PopIter<StreamReader<'a, T>>
    {
        let reader = self.read();
        PopIter {
            popped: 0,
            len: reader.len(),
            reader,
        }
    }
}

generate_pop_iterable_tuple_impl! {2}
generate_pop_iterable_tuple_impl! {3}
generate_pop_iterable_tuple_impl! {4}
generate_pop_iterable_tuple_impl! {5}
generate_pop_iterable_tuple_impl! {6}
generate_pop_iterable_tuple_impl! {7}
generate_pop_iterable_tuple_impl! {8}
generate_pop_iterable_tuple_impl! {9}
generate_pop_iterable_tuple_impl! {10}
generate_pop_iterable_tuple_impl! {11}
generate_pop_iterable_tuple_impl! {12}
generate_pop_iterable_tuple_impl! {13}
generate_pop_iterable_tuple_impl! {14}
generate_pop_iterable_tuple_impl! {15}
generate_pop_iterable_tuple_impl! {16}
generate_pop_iterable_tuple_impl! {17}
generate_pop_iterable_tuple_impl! {18}
generate_pop_iterable_tuple_impl! {19}
generate_pop_iterable_tuple_impl! {20}

impl<'a, T> Iterator for PopIter<StreamReader<'a, T>>
{
    type Item = T;

    fn next(&mut self) -> Option<Self::Item>
    {
        self.reader.pop()
    }
}

impl_iterator_for_pop_iter_tuple! {2}
impl_iterator_for_pop_iter_tuple! {3}
impl_iterator_for_pop_iter_tuple! {4}
impl_iterator_for_pop_iter_tuple! {5}
impl_iterator_for_pop_iter_tuple! {6}
impl_iterator_for_pop_iter_tuple! {7}
impl_iterator_for_pop_iter_tuple! {8}
impl_iterator_for_pop_iter_tuple! {9}
impl_iterator_for_pop_iter_tuple! {10}
impl_iterator_for_pop_iter_tuple! {11}
impl_iterator_for_pop_iter_tuple! {12}
impl_iterator_for_pop_iter_tuple! {13}
impl_iterator_for_pop_iter_tuple! {14}
impl_iterator_for_pop_iter_tuple! {15}
impl_iterator_for_pop_iter_tuple! {16}
impl_iterator_for_pop_iter_tuple! {17}
impl_iterator_for_pop_iter_tuple! {18}
impl_iterator_for_pop_iter_tuple! {19}
impl_iterator_for_pop_iter_tuple! {20}