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 RtmpProtocol;
14impl RtmpProtocol {
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!("-> RTMP {name}"))),
41 }
42}
43
44fn accept() -> ActionDefinition {
45 action("rtmp_accept", "Allow it: Rust answers with the NetConnection / NetStream success status and starts relaying", vec![], json!({"type": "rtmp_accept"}))
46}
47fn reject() -> ActionDefinition {
48 action(
49 "rtmp_reject",
50 "Refuse it: Rust answers NetConnection.Connect.Rejected, NetStream.Publish.Denied or NetStream.Play.StreamNotFound with this description",
51 vec![parameter("description", "string", "Why, as the status description the client shows, up to 256 characters", true)],
52 json!({"type": "rtmp_reject", "description": "stream key not recognised"}),
53 )
54}
55fn ignore() -> ActionDefinition {
56 action(
57 "rtmp_ignore",
58 "Acknowledge the report; nothing is sent to anyone",
59 vec![],
60 json!({"type": "rtmp_ignore"}),
61 )
62}
63pub fn send_data() -> ActionDefinition {
64 action(
65 "rtmp_send_data",
66 "Send an AMF0 data message into the stream: from a publisher's connection to every player of its stream, from a player's connection to that player (e.g. onTextData for captions, onCuePoint for markers)",
67 vec![
68 parameter("handler", "string", "The data handler name, e.g. onTextData or onCuePoint", true),
69 parameter("data", "object", "The handler's argument as a JSON object, e.g. {\"text\": \"Hello\"}", true),
70 ],
71 json!({"type": "rtmp_send_data", "handler": "onTextData", "data": {"text": "Live in five"}}),
72 )
73}
74
75pub static CONNECT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
76 EventType::new(
77 "rtmp_connect",
78 "A client completed the handshake and sends connect for an application",
79 accept().example.clone(),
80 )
81 .with_parameters(vec![
82 parameter(
83 "app",
84 "string",
85 "Application name from the connect command (e.g. live)",
86 true,
87 ),
88 parameter(
89 "tc_url",
90 "string",
91 "tcUrl the client used, e.g. rtmp://host/live",
92 false,
93 ),
94 parameter(
95 "flash_ver",
96 "string",
97 "flashVer the client announced (e.g. FMLE/3.0 or LNX 9,0,124,2)",
98 false,
99 ),
100 ])
101 .with_actions(vec![accept(), reject()])
102});
103pub static PUBLISH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
104 EventType::new(
105 "rtmp_publish",
106 "A client wants to publish a live stream (the stream name is often a stream key)",
107 accept().example.clone(),
108 )
109 .with_parameters(vec![
110 parameter("app", "string", "The connection's application", true),
111 parameter(
112 "stream",
113 "string",
114 "Stream name to publish, without query string handling",
115 true,
116 ),
117 parameter(
118 "type",
119 "string",
120 "Publishing type from the command: live, record or append",
121 true,
122 ),
123 ])
124 .with_actions(vec![accept(), reject()])
125});
126pub static PLAY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
127 EventType::new("rtmp_play", "A client wants to play a stream; if nobody publishes it yet, the player waits for the publisher", accept().example.clone())
128 .with_parameters(vec![
129 parameter("app", "string", "The connection's application", true),
130 parameter("stream", "string", "Stream name to play", true),
131 parameter("live", "boolean", "Whether someone is publishing it right now", true),
132 ])
133 .with_actions(vec![accept(), reject()])
134});
135pub static PUBLISH_ENDED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
136 EventType::new(
137 "rtmp_publish_ended",
138 "A publisher stopped; what it sent",
139 ignore().example.clone(),
140 )
141 .with_parameters(vec![
142 parameter(
143 "app",
144 "string",
145 "The application the stream was published in, e.g. live",
146 true,
147 ),
148 parameter(
149 "stream",
150 "string",
151 "The stream name (often a stream key) that was published",
152 true,
153 ),
154 parameter("duration_ms", "number", "How long it published", true),
155 parameter("video_messages", "number", "Video messages received", true),
156 parameter("audio_messages", "number", "Audio messages received", true),
157 parameter("keyframes", "number", "Video keyframes received", true),
158 parameter(
159 "media_bytes",
160 "number",
161 "Audio and video payload bytes",
162 true,
163 ),
164 parameter("players", "number", "Players attached when it ended", true),
165 ])
166 .with_actions(vec![ignore()])
167});
168
169pub fn check_answer(v: &Value) -> Result<()> {
170 ensure!(
171 crate::utils::json_budget::within_budget(v, 256 * 1024, 10_000, 16),
172 "answer exceeds the RTMP bounds"
173 );
174 match v["type"].as_str() {
175 Some("rtmp_accept" | "rtmp_ignore") => {}
176 Some("rtmp_reject") => ensure!(
177 v["description"].as_str().is_some_and(|d| !d.is_empty()
178 && d.len() <= 256
179 && !crate::utils::sanitize::has_controls(&d)),
180 "description is 1 to 256 printable characters"
181 ),
182 Some("rtmp_send_data") => {
183 ensure!(
184 v["handler"].as_str().is_some_and(|h| !h.is_empty()
185 && h.len() <= 64
186 && h.bytes()
187 .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'|' || b == b'@')),
188 "handler is a name like onTextData"
189 );
190 ensure!(v["data"].is_object(), "data is a JSON object");
191 }
192 _ => bail!("Unknown RTMP server action"),
193 }
194 Ok(())
195}
196
197fn startup(
198 name: &str,
199 kind: &str,
200 description: &str,
201 example: Value,
202 default: Option<Value>,
203) -> ParameterDefinition {
204 ParameterDefinition {
205 name: name.into(),
206 type_hint: kind.into(),
207 description: description.into(),
208 required: false,
209 example,
210 default,
211 }
212}
213
214impl Protocol for RtmpProtocol {
215 fn protocol_name(&self) -> &'static str {
216 "RTMP"
217 }
218 fn stack_name(&self) -> &'static str {
219 "ETH>IP>TCP>RTMP"
220 }
221 fn description(&self) -> &'static str {
222 "RTMP live server: publishers push streams, players pull them; the handler admits each and can inject data messages"
223 }
224 fn keywords(&self) -> Vec<&'static str> {
225 vec![
226 "rtmp",
227 "flash",
228 "live streaming",
229 "obs",
230 "ffmpeg",
231 "publish",
232 "ingest",
233 ]
234 }
235 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
236 vec![send_data()]
237 }
238 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
239 vec![accept(), reject(), ignore()]
240 }
241 fn get_event_types(&self) -> Vec<EventType> {
242 vec![
243 CONNECT_EVENT.clone(),
244 PUBLISH_EVENT.clone(),
245 PLAY_EVENT.clone(),
246 PUBLISH_ENDED_EVENT.clone(),
247 ]
248 }
249 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
250 vec![startup("idle_timeout_secs", "integer", "Seconds a connection may send nothing before it is closed (players receiving media are kept)", json!(30), Some(json!(super::IDLE_TIMEOUT.as_secs())))]
251 }
252 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
253 use crate::protocol::metadata::*;
254 ProtocolMetadataV2::builder()
255 .state(DevelopmentState::Experimental)
256 .privilege_requirement(PrivilegeRequirement::None)
257 .well_known_port(1935)
258 .implementation("Hand-written RTMP: simple handshake, chunk streams (all header formats, extended timestamps, chunk size, acknowledgements), AMF0 commands, NetConnection/NetStream flow, live relay with cached metadata and sequence headers, keyframe join and per-player timestamp rebasing")
259 .llm_control("Which applications, publishers (stream keys) and players are admitted, and data messages pushed into streams")
260 .e2e_testing("tests/server/rtmp: FFmpeg (independent) publishes H.264/AAC and plays it back decoding frames; MediaMTX (independent Go RTMP) pulls the stream and reports its tracks; refused apps and stream keys")
261 .notes("Live only (no recording or VOD), AMF0 only, no RTMPS/RTMPT, no complex (digest) handshake, no enhanced-RTMP multitrack. Media is relayed, never stored.")
262 .answers_on_failure()
263 .max_inbound_bytes(super::chunk::MAX_MESSAGE)
264 .build()
265 }
266 fn example_prompt(&self) -> &'static str {
267 "RTMP ingest on port 1935 that accepts the stream key abc123 on app live and lets anyone watch"
268 }
269 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
270 let llm = json!({"type":"open_server","base_stack":"rtmp","port":1935,"instruction":"Accept app live; only stream key abc123 may publish; anyone may play"});
271 let mut static_example = llm.clone();
272 static_example["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"static","actions":[{"type":"rtmp_accept"}]}}]);
273 let mut scripted = llm.clone();
274 scripted["event_handlers"] = json!([
275 {"event_pattern":"rtmp_publish","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'rtmp_accept'} if e['stream']=='abc123' else {'type':'rtmp_reject','description':'bad stream key'}]}))"}},
276 {"event_pattern":"*","handler":{"type":"static","actions":[{"type":"rtmp_accept"}]}}
277 ]);
278 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
279 }
280 fn group_name(&self) -> &'static str {
281 "Real-time"
282 }
283}
284
285impl Server for RtmpProtocol {
286 fn spawn(
287 &self,
288 ctx: SpawnContext,
289 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
290 {
291 Box::pin(super::spawn(ctx))
292 }
293 fn execute_action(&self, v: Value) -> Result<ActionResult> {
294 check_answer(&v)?;
295 Ok(ActionResult::Custom {
296 name: v["type"].as_str().unwrap_or_default().into(),
297 data: v,
298 })
299 }
300}