appam 0.1.1

High-throughput, traceable, reliable Rust agent framework for long-horizon AI sessions and easy extensibility
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112

//! Channel consumer for forwarding events to mpsc channels.

use anyhow::Result;
use tokio::sync::mpsc;

use crate::agent::streaming::{StreamConsumer, StreamEvent};

/// Consumer that forwards events to a tokio mpsc channel.
///
/// Useful for bridging agent execution with async streams, web sockets,
/// SSE responses, or any other channel-based communication.
///
/// # Examples
///
/// ```ignore
/// use appam::agent::consumers::ChannelConsumer;
/// use tokio::sync::mpsc;
///
/// let (tx, mut rx) = mpsc::unbounded_channel();
/// let consumer = ChannelConsumer::new(tx);
///
/// // In another task:
/// tokio::spawn(async move {
///     while let Some(event) = rx.recv().await {
///         println!("Event: {:?}", event);
///     }
/// });
/// ```
pub struct ChannelConsumer {
    tx: mpsc::UnboundedSender<StreamEvent>,
}

impl ChannelConsumer {
    /// Create a new channel consumer.
    ///
    /// Events are sent to the provided channel. If the channel is closed
    /// (receiver dropped), subsequent sends will fail silently.
    pub fn new(tx: mpsc::UnboundedSender<StreamEvent>) -> Self {
        Self { tx }
    }
}

impl StreamConsumer for ChannelConsumer {
    fn on_event(&self, event: &StreamEvent) -> Result<()> {
        // Clone the event and send to channel
        // If send fails (receiver dropped), we silently ignore it
        // as this is not a critical error
        let _ = self.tx.send(event.clone());
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_channel_consumer() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let consumer = ChannelConsumer::new(tx);

        let event = StreamEvent::Content {
            content: "test".to_string(),
        };

        consumer.on_event(&event).unwrap();

        let received = rx.recv().await.unwrap();
        match received {
            StreamEvent::Content { content } => assert_eq!(content, "test"),
            _ => panic!("Wrong event type"),
        }
    }

    #[tokio::test]
    async fn test_channel_consumer_multiple_events() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let consumer = ChannelConsumer::new(tx);

        consumer
            .on_event(&StreamEvent::SessionStarted {
                session_id: "test".to_string(),
            })
            .unwrap();

        consumer
            .on_event(&StreamEvent::Content {
                content: "hello".to_string(),
            })
            .unwrap();

        consumer.on_event(&StreamEvent::Done).unwrap();

        // Should receive all 3 events
        assert!(rx.recv().await.is_some());
        assert!(rx.recv().await.is_some());
        assert!(rx.recv().await.is_some());
    }

    #[test]
    fn test_channel_consumer_closed_receiver() {
        let (tx, rx) = mpsc::unbounded_channel();
        let consumer = ChannelConsumer::new(tx);

        // Drop receiver
        drop(rx);

        // Should not panic, just silently ignore
        let result = consumer.on_event(&StreamEvent::Done);
        assert!(result.is_ok());
    }
}