1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, ParameterDefinition,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::rtmp::actions::{action, parameter};
8use anyhow::{bail, ensure, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct RtmpClientProtocol;
14impl RtmpClientProtocol {
15 pub fn new() -> Self {
16 Self
17 }
18}
19
20fn play() -> ActionDefinition {
21 action(
22 "rtmp_play",
23 "Play a stream for a while and report what arrived: status codes, metadata, codecs, message and keyframe counts, timestamps and data messages",
24 vec![
25 parameter("stream", "string", "Stream name in the connected app, e.g. cam1", true),
26 parameter("seconds", "number", "How long to watch, 1 to 60 (default 5)", false),
27 ],
28 json!({"type": "rtmp_play", "stream": "cam1", "seconds": 5}),
29 )
30}
31fn publish() -> ActionDefinition {
32 action(
33 "rtmp_publish",
34 "Publish an FLV file the operator supplied as a live stream, its tags paced by their timestamps, then unpublish and report",
35 vec![
36 parameter("stream", "string", "Stream name (often a stream key) to publish as", true),
37 parameter("flv_file", "string", "Path of an FLV file under the client's media_root directory (64 MiB at most); a relative path is resolved against that directory, and nothing outside it is read", true),
38 parameter("realtime", "boolean", "Pace tags by their timestamps (default true); false sends as fast as possible", false),
39 ],
40 json!({"type": "rtmp_publish", "stream": "abc123", "flv_file": "clip.flv"}),
41 )
42}
43fn disconnect() -> ActionDefinition {
44 action(
45 "disconnect",
46 "Close the connection",
47 vec![],
48 json!({"type": "disconnect"}),
49 )
50}
51fn actions() -> Vec<ActionDefinition> {
52 vec![play(), publish(), disconnect()]
53}
54
55pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
56 EventType::new(
57 "rtmp_connected",
58 "The server accepted connect for the application",
59 play().example.clone(),
60 )
61 .with_parameters(vec![
62 parameter("app", "string", "The application connected to", true),
63 parameter(
64 "server",
65 "object",
66 "The server's properties from _result (fmsVer, capabilities)",
67 true,
68 ),
69 ])
70 .with_actions(actions())
71});
72pub static REPORT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
73 EventType::new(
74 "rtmp_report",
75 "What a play or publish did",
76 disconnect().example.clone(),
77 )
78 .with_parameters(vec![
79 parameter("operation", "string", "play or publish", true),
80 parameter(
81 "stream",
82 "string",
83 "The stream name that was played or published",
84 true,
85 ),
86 parameter(
87 "status_codes",
88 "array",
89 "onStatus codes the server sent, e.g. NetStream.Play.Start",
90 true,
91 ),
92 parameter(
93 "video_messages",
94 "number",
95 "Video messages received (play) or sent (publish)",
96 true,
97 ),
98 parameter(
99 "audio_messages",
100 "number",
101 "Audio messages received or sent",
102 true,
103 ),
104 parameter("keyframes", "number", "Video keyframes", true),
105 parameter(
106 "video_codec",
107 "string",
108 "avc, hevc, av1, vp9 or the FLV codec ID",
109 false,
110 ),
111 parameter(
112 "audio_codec",
113 "string",
114 "aac, mp3 or the FLV sound format",
115 false,
116 ),
117 parameter(
118 "metadata",
119 "object",
120 "onMetaData received or published",
121 false,
122 ),
123 parameter(
124 "first_timestamp",
125 "number",
126 "First media timestamp, ms",
127 false,
128 ),
129 parameter(
130 "last_timestamp",
131 "number",
132 "Last media timestamp, ms",
133 false,
134 ),
135 parameter(
136 "data_messages",
137 "array",
138 "Other data messages received: [{handler, data}]",
139 false,
140 ),
141 ])
142 .with_actions(actions())
143});
144
145impl Protocol for RtmpClientProtocol {
146 fn protocol_name(&self) -> &'static str {
147 "RTMP"
148 }
149 fn stack_name(&self) -> &'static str {
150 "ETH>IP>TCP>RTMP"
151 }
152 fn keywords(&self) -> Vec<&'static str> {
153 vec!["rtmp", "live streaming", "play", "publish", "ingest"]
154 }
155 fn description(&self) -> &'static str {
156 "RTMP client: connects to an application, plays streams and reports what arrives, publishes FLV files live"
157 }
158 fn get_async_actions(&self, _: &crate::state::app_state::AppState) -> Vec<ActionDefinition> {
159 actions()
160 }
161 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
162 vec![]
163 }
164 fn get_event_types(&self) -> Vec<EventType> {
165 vec![CONNECTED_EVENT.clone(), REPORT_EVENT.clone()]
166 }
167 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
168 vec![
169 ParameterDefinition {
170 name: "app".into(),
171 type_hint: "string".into(),
172 description: "Application to connect to, e.g. live".into(),
173 required: false,
174 example: json!("live"),
175 default: Some(json!(super::DEFAULT_APP)),
176 },
177 ParameterDefinition {
178 name: crate::client::media_root::MEDIA_ROOT_PARAM.into(),
179 type_hint: "string".into(),
180 description: "Directory rtmp_publish may read FLV files from; nothing outside it is sent. Defaults to NetGet's own media directory under the platform's local-data dir.".into(),
181 required: false,
182 example: json!("/var/lib/netget/media"),
183 default: None,
184 },
185 ]
186 }
187 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
188 use crate::protocol::metadata::*;
189 ProtocolMetadataV2::builder()
190 .state(DevelopmentState::Experimental)
191 .privilege_requirement(PrivilegeRequirement::None)
192 .implementation("The server's handshake, chunk and AMF0 code as a client; FLV demuxing for publish")
193 .llm_control("Which streams to play or publish and when")
194 .e2e_testing("tests/client/rtmp: MediaMTX (independent Go RTMP) serves a stream FFmpeg publishes, which the client plays; the client publishes an FLV that MediaMTX reports with its tracks")
195 .notes("AMF0 only, no RTMPS, simple handshake; publishes FLV files only.")
196 .max_inbound_bytes(crate::server::rtmp::chunk::MAX_MESSAGE)
197 .build()
198 }
199 fn example_prompt(&self) -> &'static str {
200 "Connect to rtmp://127.0.0.1:1935/live and watch stream cam1 for five seconds"
201 }
202 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
203 let llm = json!({"type":"open_client","protocol":"rtmp","remote_addr":"127.0.0.1:1935","instruction":"Watch cam1 for five seconds","startup_params":{"app":"live"}});
204 let mut static_example = llm.clone();
205 static_example["event_handlers"] = json!([
206 {"event_pattern":"rtmp_connected","handler":{"type":"static","actions":[play().example]}},
207 {"event_pattern":"rtmp_report","handler":{"type":"static","actions":[{"type":"disconnect"}]}}
208 ]);
209 let mut scripted = static_example.clone();
210 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'disconnect'}] if e['video_messages']>0 else [{'type':'rtmp_play','stream':e['stream'],'seconds':5}]}))"});
211 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
212 }
213 fn group_name(&self) -> &'static str {
214 "Real-time"
215 }
216}
217
218fn name_ok(s: Option<&str>) -> bool {
219 s.is_some_and(|s| !s.is_empty() && s.len() <= 256 && !crate::utils::sanitize::has_controls(&s))
220}
221
222impl Client for RtmpClientProtocol {
223 fn connect(
224 &self,
225 ctx: ConnectContext,
226 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
227 {
228 Box::pin(super::connect(ctx))
229 }
230 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
231 match v["type"].as_str() {
232 Some("rtmp_play") => {
233 ensure!(name_ok(v["stream"].as_str()), "stream is a name");
234 if let Some(s) = v.get("seconds").filter(|s| !s.is_null()) {
235 ensure!(
236 s.as_f64().is_some_and(|s| (1.0..=60.0).contains(&s)),
237 "seconds is 1 to 60"
238 );
239 }
240 }
241 Some("rtmp_publish") => {
242 ensure!(name_ok(v["stream"].as_str()), "stream is a name");
243 ensure!(name_ok(v["flv_file"].as_str()), "flv_file is a path");
244 }
245 Some("disconnect") => return Ok(ClientActionResult::Disconnect),
246 _ => bail!("Unknown RTMP client action"),
247 }
248 Ok(ClientActionResult::Custom {
249 name: v["type"].as_str().unwrap_or_default().into(),
250 data: v,
251 })
252 }
253}