futuresdr 0.0.41

use std::pin::Pin;
use std::time::Duration;
use web_time::Instant;

use crate::runtime::Timer;
use crate::runtime::dev::prelude::*;

/// Output the same message periodically.
#[derive(Block)]
#[message_outputs(out)]
pub struct MessageSource {
    message: Pmt,
    interval: Duration,
    t_last: Instant,
    n_messages: Option<usize>,
    timer: Option<Timer>,
}

impl MessageSource {
    /// Create MessageSource block
    pub fn new(message: Pmt, interval: Duration, n_messages: Option<usize>) -> Self {
        Self {
            message,
            interval,
            t_last: Instant::now(),
            n_messages,
            timer: None,
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[doc(hidden)]
impl Kernel for MessageSource {
    type BlockOn = Timer;

    fn block_on(&mut self) -> Option<Pin<&mut Timer>> {
        self.timer.as_mut().map(Pin::new)
    }

    async fn work(
        &mut self,
        io: &mut WorkIo,
        mo: &mut MessageOutputs,
        _b: &mut BlockMeta,
    ) -> Result<()> {
        let now = Instant::now();

        if now >= self.t_last + self.interval {
            mo.post("out", self.message.clone()).await?;
            self.t_last = now;
            if let Some(ref mut n) = self.n_messages {
                *n -= 1;
                if *n == 0 {
                    io.finished = true;
                }
            }
        }

        self.timer = Some(Timer::after(self.t_last + self.interval - Instant::now()));
        io.block_on();

        Ok(())
    }

    async fn init(&mut self, _mo: &mut MessageOutputs, _b: &mut BlockMeta) -> Result<()> {
        self.t_last = Instant::now();
        Ok(())
    }
}

#[cfg(target_arch = "wasm32")]
#[doc(hidden)]
impl Kernel for MessageSource {
    type BlockOn = Timer;

    fn block_on(&mut self) -> Option<Pin<&mut Timer>> {
        self.timer.as_mut().map(Pin::new)
    }

    async fn work(
        &mut self,
        io: &mut WorkIo,
        mo: &mut MessageOutputs,
        _b: &mut BlockMeta,
    ) -> Result<()> {
        let now = Instant::now();

        if now >= self.t_last + self.interval {
            mo.post("out", self.message.clone()).await?;
            self.t_last = now;
            if let Some(ref mut n) = self.n_messages {
                *n -= 1;
                if *n == 0 {
                    io.finished = true;
                }
            }
        }

        self.timer = Some(Timer::after(self.t_last + self.interval - Instant::now()));
        io.block_on();

        Ok(())
    }

    async fn init(&mut self, _mo: &mut MessageOutputs, _b: &mut BlockMeta) -> Result<()> {
        self.t_last = Instant::now();
        Ok(())
    }
}

/// Repeats a fixed message on an interval
///
/// # Message Inputs
///
/// No message inputs.
///
/// # Message Outputs
///
/// `out`: Repeated message output.
///
/// # Usage
/// ```
/// use std::time;
/// use futuresdr::blocks::MessageSourceBuilder;
/// use futuresdr::runtime::{Flowgraph, Pmt};
///
/// let mut fg = Flowgraph::new();
///
/// // Repeat the message "foo" every 100ms twenty times
/// let msg_source = fg.add(
///     MessageSourceBuilder::new(
///         Pmt::String("foo".to_string()),
///         time::Duration::from_millis(100),
///     )
///     .n_messages(20)
///     .build()
/// );
/// ```
pub struct MessageSourceBuilder {
    message: Pmt,
    duration: Duration,
    n_messages: Option<usize>,
}

impl MessageSourceBuilder {
    /// Create MessageSource builder
    pub fn new(message: Pmt, duration: Duration) -> MessageSourceBuilder {
        MessageSourceBuilder {
            message,
            duration,
            n_messages: None,
        }
    }
    /// Number of message to send
    #[must_use]
    pub fn n_messages(mut self, n: usize) -> MessageSourceBuilder {
        self.n_messages = Some(n);
        self
    }
    /// Build Message Source block
    pub fn build(self) -> MessageSource {
        MessageSource::new(self.message, self.duration, self.n_messages)
    }
}