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
88
89
90
91
92
93
94
95
96
97
98

use crate::stream::{CloneableStreamable, StreamMessage, Streamable};
use actix::prelude::*; // Changed to prelude

// DropWhileActor drops elements from the stream while the predicate is true,
// then passes through all subsequent elements.
#[derive(Debug)]
pub(crate) struct DropWhileActor<Out, P>
where
    Out: Streamable,
    P: FnMut(&Out) -> bool + Send + 'static + Unpin,
{
    predicate: P,
    downstream: Recipient<StreamMessage<Out>>,
    is_dropping: bool,
}

impl<Out, P> DropWhileActor<Out, P>
where
    Out: Streamable,
    P: FnMut(&Out) -> bool + Send + 'static + Unpin,
{
    pub(crate) fn new(predicate: P, downstream: Recipient<StreamMessage<Out>>) -> Self {
        Self {
            predicate,
            downstream,
            is_dropping: true,
        }
    }
}

impl<Out, P> Actor for DropWhileActor<Out, P>
where
    Out: Streamable,
    P: FnMut(&Out) -> bool + Send + 'static + Unpin,
{
    type Context = Context<Self>;

    fn started(&mut self, _ctx: &mut Self::Context) {
        // log::debug!("DropWhileActor started. Initial state: is_dropping = {}", self.is_dropping);
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        // log::debug!("DropWhileActor stopped. Ensuring End message is sent to downstream.");
        let _ = self.downstream.try_send(StreamMessage::End);
    }
}

impl<Out, P> Handler<StreamMessage<Out>> for DropWhileActor<Out, P>
where
    Out: CloneableStreamable,
    P: FnMut(&Out) -> bool + Send + 'static + Unpin,
{
    type Result = ();

    fn handle(&mut self, msg: StreamMessage<Out>, 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(item) => {
                if self.is_dropping {
                    if !(self.predicate)(&item) {
                        self.is_dropping = false;
                        if self
                            .downstream
                            .try_send(StreamMessage::Element(item.clone()))
                            .is_err()
                        {
                            if ctx.state() != ActorState::Stopping
                                && ctx.state() != ActorState::Stopped
                            {
                                ctx.stop();
                            }
                        }
                    }
                } else {
                    if self
                        .downstream
                        .try_send(StreamMessage::Element(item))
                        .is_err()
                    {
                        if ctx.state() != ActorState::Stopping && ctx.state() != ActorState::Stopped
                        {
                            ctx.stop();
                        }
                    }
                }
            }
            StreamMessage::End => {
                if ctx.state() != ActorState::Stopping && ctx.state() != ActorState::Stopped {
                    ctx.stop();
                }
            }
        }
    }
}