1use crate::llm::actions::{
2 protocol_trait::{ActionResult, Protocol, Server},
3 ActionDefinition, Parameter, ParameterDefinition,
4};
5use crate::protocol::log_template::LogTemplate;
6use crate::protocol::{EventType, SpawnContext};
7use crate::state::app_state::AppState;
8use anyhow::{bail, ensure, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct SrtProtocol;
14impl SrtProtocol {
15 pub fn new() -> Self {
16 Self
17 }
18}
19
20pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
21 Parameter {
22 name: name.into(),
23 type_hint: kind.into(),
24 description: description.into(),
25 required,
26 }
27}
28
29pub fn action(
30 name: &str,
31 description: &str,
32 parameters: Vec<Parameter>,
33 example: Value,
34) -> ActionDefinition {
35 ActionDefinition {
36 name: name.into(),
37 description: description.into(),
38 parameters,
39 example,
40 log_template: Some(LogTemplate::new().with_info(format!("-> SRT {name}"))),
41 }
42}
43
44fn accept() -> ActionDefinition {
45 action("srt_accept", "Admit the caller: Rust completes the SRT handshake; a publisher's stream is relayed to every reader of the same resource", vec![], json!({"type": "srt_accept"}))
46}
47fn reject() -> ActionDefinition {
48 action(
49 "srt_reject",
50 "Refuse the caller with an SRT server reject code in the handshake",
51 vec![parameter("reason", "string", "unauthorized (2401), forbidden (2403), not_found (2404), bad_request (2400), conflict (2409) or overload (2402)", true)],
52 json!({"type": "srt_reject", "reason": "unauthorized"}),
53 )
54}
55fn ignore() -> ActionDefinition {
56 action(
57 "srt_ignore",
58 "Acknowledge the report; nothing is sent to anyone",
59 vec![],
60 json!({"type": "srt_ignore"}),
61 )
62}
63pub fn send_text() -> ActionDefinition {
64 action(
65 "srt_send_text",
66 "Send a UTF-8 text message to this reader as one SRT message (alongside any relayed stream)",
67 vec![parameter("text", "string", "The message, up to 1316 bytes so it fits one SRT packet", true)],
68 json!({"type": "srt_send_text", "text": "stream starts in one minute"}),
69 )
70}
71
72pub static CONNECT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
73 EventType::new(
74 "srt_connect",
75 "A caller completed the SRT induction and asks to connect with this stream ID",
76 accept().example.clone(),
77 )
78 .with_parameters(vec![
79 parameter(
80 "stream_id",
81 "string",
82 "The stream ID as sent, e.g. #!::r=live/cam,m=publish or publish:live/cam",
83 true,
84 ),
85 parameter(
86 "resource",
87 "string",
88 "The resource it names, e.g. live/cam",
89 true,
90 ),
91 parameter(
92 "mode",
93 "string",
94 "publish (sends a stream) or request (reads one)",
95 true,
96 ),
97 parameter("user", "string", "The user (u=) it names, when any", false),
98 parameter("remote", "string", "The caller's UDP address", true),
99 ])
100 .with_actions(vec![accept(), reject()])
101});
102pub static CLOSED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
103 EventType::new(
104 "srt_closed",
105 "An admitted connection ended, with the SRT statistics of its last report",
106 ignore().example.clone(),
107 )
108 .with_parameters(vec![
109 parameter(
110 "resource",
111 "string",
112 "The resource it published or read",
113 true,
114 ),
115 parameter("mode", "string", "publish or request", true),
116 parameter("duration_ms", "number", "How long it was connected", true),
117 parameter("reason", "string", "Why it ended: the caller closed, the publisher went idle, a slow reader was dropped, or the operator disconnected it", true),
118 parameter(
119 "statistics",
120 "object",
121 "rx/tx packets and bytes, lost, retransmitted and dropped packets, RTT",
122 false,
123 ),
124 ])
125 .with_actions(vec![ignore()])
126});
127
128pub fn check_answer(v: &Value) -> Result<()> {
129 match v["type"].as_str() {
130 Some("srt_accept" | "srt_ignore") => {}
131 Some("srt_reject") => {
132 let r = v["reason"].as_str().unwrap_or_default();
133 ensure!(
134 super::REJECT_CODES.iter().any(|(n, _)| *n == r),
135 "unknown reject reason {r:?}"
136 );
137 }
138 Some("srt_send_text") => ensure!(
139 v["text"]
140 .as_str()
141 .is_some_and(|t| !t.is_empty() && t.len() <= 1316),
142 "text is 1 to 1316 bytes"
143 ),
144 _ => bail!("Unknown SRT server action"),
145 }
146 Ok(())
147}
148
149fn startup(
150 name: &str,
151 kind: &str,
152 description: &str,
153 example: Value,
154 default: Option<Value>,
155) -> ParameterDefinition {
156 ParameterDefinition {
157 name: name.into(),
158 type_hint: kind.into(),
159 description: description.into(),
160 required: false,
161 example,
162 default,
163 }
164}
165
166impl Protocol for SrtProtocol {
167 fn protocol_name(&self) -> &'static str {
168 "SRT"
169 }
170 fn stack_name(&self) -> &'static str {
171 "ETH>IP>UDP>SRT"
172 }
173 fn description(&self) -> &'static str {
174 "SRT listener: callers publish or read streams by stream ID; Rust relays each publisher to its readers over SRT's retransmission and latency buffer"
175 }
176 fn keywords(&self) -> Vec<&'static str> {
177 vec![
178 "srt",
179 "secure reliable transport",
180 "haivision",
181 "contribution",
182 "live streaming",
183 "mpeg-ts",
184 ]
185 }
186 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
187 vec![send_text()]
188 }
189 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
190 vec![accept(), reject(), ignore()]
191 }
192 fn get_event_types(&self) -> Vec<EventType> {
193 vec![CONNECT_EVENT.clone(), CLOSED_EVENT.clone()]
194 }
195 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
196 vec![
197 startup("latency_ms", "integer", "SRT latency (TSBPD delay) in milliseconds, 20 to 8000", json!(200), Some(json!(super::DEFAULT_LATENCY.as_millis() as u64))),
198 startup("idle_timeout_secs", "integer", "Seconds a publisher may send nothing before it is closed", json!(5), Some(json!(super::IDLE_TIMEOUT.as_secs()))),
199 startup("passphrase", "string", "Encrypt with AES-128 using this passphrase (10 to 79 characters); unencrypted without it", json!("correct horse battery"), None),
200 ]
201 }
202 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
203 use crate::protocol::metadata::*;
204 ProtocolMetadataV2::builder()
205 .port_transport(PortTransport::Udp)
206 .state(DevelopmentState::Experimental)
207 .privilege_requirement(PrivilegeRequirement::None)
208 .implementation("srt-tokio 0.4.4 (pure-Rust SRT: handshake, ARQ, TSBPD, too-late drop, AES); stream-ID access-control parsing and the publish-to-readers relay are NetGet's")
209 .llm_control("Which callers may publish or read which resources, and text messages to readers")
210 .e2e_testing("tests/server/srt: libsrt 1.5 (srt-live-transmit, independent) publishes a paced H.264/AAC MPEG-TS stream and a second libsrt caller reads it back, which ffprobe decodes; a forbidden resource is refused in the handshake. MediaMTX (gosrt) refuses srt-tokio's SRT 1.3 handshake in both directions, so it is not a peer")
211 .notes("Live mode only, single publisher per resource, no rendezvous, no bidirectional mode, no FEC or bonding. The relay forwards payloads; it does not parse MPEG-TS. srt-tokio announces SRT 1.3 and puts its payload size in the handshake MSS field; the payload size is raised to 1456 so libsrt peers can send 1316-byte messages.")
212 .answers_on_failure()
213 .max_inbound_bytes(1500)
214 .build()
215 }
216 fn example_prompt(&self) -> &'static str {
217 "SRT ingest on UDP port 9000 that admits publishers to live/* and anyone reading"
218 }
219 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
220 let llm = json!({"type":"open_server","base_stack":"srt","port":9000,"instruction":"Admit publishers to resources under live/ and every reader"});
221 let mut static_example = llm.clone();
222 static_example["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"static","actions":[{"type":"srt_accept"}]}}]);
223 let mut scripted = llm.clone();
224 scripted["event_handlers"] = json!([
225 {"event_pattern":"srt_connect","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nok=e['mode']=='request' or e['resource'].startswith('live/')\nprint(json.dumps({'actions':[{'type':'srt_accept'} if ok else {'type':'srt_reject','reason':'forbidden'}]}))"}},
226 {"event_pattern":"*","handler":{"type":"static","actions":[{"type":"srt_ignore"}]}}
227 ]);
228 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
229 }
230 fn group_name(&self) -> &'static str {
231 "Real-time"
232 }
233}
234
235impl Server for SrtProtocol {
236 fn spawn(
237 &self,
238 ctx: SpawnContext,
239 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
240 {
241 Box::pin(super::spawn(ctx))
242 }
243 fn execute_action(&self, v: Value) -> Result<ActionResult> {
244 check_answer(&v)?;
245 Ok(ActionResult::Custom {
246 name: v["type"].as_str().unwrap_or_default().into(),
247 data: v,
248 })
249 }
250}