vuo 0.1.1

An asynchronous stream processing library for Rust, built on Actix, providing a rich set of functional operators.
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

use crate::stream::{StreamMessage, Streamable, CloneableStreamable};
use actix::prelude::*; // Changed to prelude
use std::marker::PhantomData;
// Removed: use std::fmt;

#[derive(Debug)] // Added Debug for the actor struct itself
pub(crate) struct MappingActor<In, Out, F>
where
    In: Streamable,          // Input type, requires Streamable
    Out: CloneableStreamable, // Output type, requires CloneableStreamable for messages
    F: FnMut(In) -> Out + Send + 'static + Clone + Unpin, // Mapping function
{
    map_fn: F,
    downstream: Recipient<StreamMessage<Out>>,
    _phantom_in: PhantomData<In>, // To hold the In type
}

impl<In, Out, F> MappingActor<In, Out, F>
where
    In: Streamable,
    Out: CloneableStreamable,
    F: FnMut(In) -> Out + Send + 'static + Clone + Unpin,
{
    pub(crate) fn new(map_fn: F, downstream: Recipient<StreamMessage<Out>>) -> Self {
        MappingActor {
            map_fn,
            downstream,
            _phantom_in: PhantomData,
        }
    }
}

impl<In, Out, F> Actor for MappingActor<In, Out, F>
where
    In: Streamable,
    Out: CloneableStreamable,
    F: FnMut(In) -> Out + Send + 'static + Clone + Unpin,
{
    type Context = Context<Self>;

    fn started(&mut self, _ctx: &mut Self::Context) {
        // log::debug!("MappingActor started.");
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        // log::debug!("MappingActor stopped, sending End to downstream.");
        // Ensure downstream is notified if the actor stops for any reason.
        let _ = self.downstream.try_send(StreamMessage::End);
    }
}

impl<In, Out, F> Handler<StreamMessage<In>> for MappingActor<In, Out, F>
where
    In: Streamable, // In is moved into map_fn
    Out: CloneableStreamable, // Out is produced and sent
    F: FnMut(In) -> Out + Send + 'static + Clone + Unpin,
{
    type Result = ();

    fn handle(&mut self, msg: StreamMessage<In>, ctx: &mut Context<Self>) {
        // If actor is already stopping/stopped, ignore new messages.
        if ctx.state() == ActorState::Stopping || ctx.state() == ActorState::Stopped {
            return;
        }

        match msg {
            StreamMessage::Element(elem) => {
                // log::trace!("MappingActor received element: {:?}", elem);
                let mapped_elem = (self.map_fn)(elem); // elem is moved here
                // log::trace!("MappingActor mapped to: {:?}", mapped_elem);
                if self.downstream.try_send(StreamMessage::Element(mapped_elem)).is_err() {
                    // log::warn!("MappingActor: Downstream recipient is closed. Stopping actor.");
                    if ctx.state() != ActorState::Stopping && ctx.state() != ActorState::Stopped {
                        ctx.stop(); // Should work now
                    }
                }
            }
            StreamMessage::End => {
                // log::debug!("MappingActor received End. Stopping actor.");
                // Upstream ended. The `stopped` method will ensure downstream gets an End message.
                if ctx.state() != ActorState::Stopping && ctx.state() != ActorState::Stopped {
                    ctx.stop(); // Should work now
                }
            }
        }
    }
}