ag_ui/client/transport/
mod.rs

1//! Getting events from somewhere.
2//!
3//! Everything else in this crate is synchronous. This is the one layer that
4//! talks to the outside world, and the only place `async` appears — which is
5//! what lets the rest of the crate run under any executor, or none.
6//!
7//! [`Transport`] is deliberately small: hand it a [`RunAgentInput`](https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/input/struct.RunAgentInput.html), get back a
8//! stream of [`Event`](https://kimsoungryoul.github.io/ag-ui-rust/api/ag_ui/event/enum.Event.html)s. Implementations shipped here:
9//!
10//! - [`sse`] — the `text/event-stream` decoder every HTTP transport needs.
11#![cfg_attr(
12    feature = "http",
13    doc = "- [`http`] *(feature `http`)* — [`HttpTransport`], backed by `reqwest`."
14)]
15#![cfg_attr(
16    not(feature = "http"),
17    doc = "- `http` *(feature `http`, off in this build)* — `HttpTransport`, backed by `reqwest`."
18)]
19//! - [`replay`] — [`ReplayTransport`], which serves a scripted list of events
20//!   and records what was sent to it. Tests use it; so does the doc example.
21//!
22//! A wasm frontend, an in-process agent, a websocket, a recorded fixture: each
23//! is an `impl Transport`, and nothing above this module changes.
24
25pub mod replay;
26pub mod sse;
27
28#[cfg(feature = "http")]
29pub mod http;
30
31use std::future::Future;
32use std::pin::Pin;
33use std::sync::Arc;
34
35use crate::{Event, RunAgentInput};
36use futures_core::Stream;
37
38use crate::client::error::Result;
39
40pub use replay::ReplayTransport;
41pub use sse::{SseDecoder, SseFrame, decode_events};
42
43#[cfg(feature = "http")]
44pub use http::{HttpTransport, HttpTransportBuilder};
45
46/// A boxed stream of events, as a transport hands it over.
47///
48/// `Send` everywhere except wasm, where the browser APIs a transport would be
49/// built on are single-threaded and not `Send` at all. Requiring it there would
50/// make the wasm case — the reason this crate abstracts the transport in the
51/// first place — impossible to satisfy.
52#[cfg(not(target_family = "wasm"))]
53pub type EventStream = Pin<Box<dyn Stream<Item = Result<Event>> + Send>>;
54
55/// A boxed stream of events, as a transport hands it over.
56#[cfg(target_family = "wasm")]
57pub type EventStream = Pin<Box<dyn Stream<Item = Result<Event>>>>;
58
59/// The future [`Transport::run`] returns: connecting, before any event arrives.
60#[cfg(not(target_family = "wasm"))]
61pub type TransportFuture = Pin<Box<dyn Future<Output = Result<EventStream>> + Send>>;
62
63/// The future [`Transport::run`] returns: connecting, before any event arrives.
64#[cfg(target_family = "wasm")]
65pub type TransportFuture = Pin<Box<dyn Future<Output = Result<EventStream>>>>;
66
67/// Somewhere an agent's events come from.
68///
69/// # Why the future is `'static`
70///
71/// A transport is usually held inside a [`Session`](crate::client::Session), which
72/// mutates its own state as events arrive. If the returned future borrowed the
73/// transport, that borrow would live as long as the run and the session could
74/// not touch itself while streaming. So `run` clones what it needs —
75/// `reqwest::Client` is explicitly designed for exactly that — and the future
76/// stands alone.
77pub trait Transport {
78    /// Starts a run and connects to its event stream.
79    ///
80    /// Failing to connect is an error from the future; failing mid-stream is an
81    /// error item in the stream.
82    fn run(&self, input: RunAgentInput) -> TransportFuture;
83}
84
85impl<T: Transport + ?Sized> Transport for &T {
86    fn run(&self, input: RunAgentInput) -> TransportFuture {
87        (**self).run(input)
88    }
89}
90
91impl<T: Transport + ?Sized> Transport for Box<T> {
92    fn run(&self, input: RunAgentInput) -> TransportFuture {
93        (**self).run(input)
94    }
95}
96
97impl<T: Transport + ?Sized> Transport for Arc<T> {
98    fn run(&self, input: RunAgentInput) -> TransportFuture {
99        (**self).run(input)
100    }
101}
102
103/// Boxes a stream into the shape [`Transport::run`] returns.
104#[cfg(not(target_family = "wasm"))]
105pub fn boxed_stream(stream: impl Stream<Item = Result<Event>> + Send + 'static) -> EventStream {
106    Box::pin(stream)
107}
108
109/// Boxes a stream into the shape [`Transport::run`] returns.
110#[cfg(target_family = "wasm")]
111pub fn boxed_stream(stream: impl Stream<Item = Result<Event>> + 'static) -> EventStream {
112    Box::pin(stream)
113}
ag_ui/client/transport/mod.rs

ag_ui/client/transport/
mod.rs
