ez_ffmpeg/rtmp/
embed_rtmp_server.rs

1use crate::core::context::output::Output;
2use crate::error::Error::{RtmpCreateStream, RtmpStreamAlreadyExists};
3use crate::flv::flv_buffer::FlvBuffer;
4use crate::flv::flv_tag::FlvTag;
5use crate::rtmp::reactor::{effective_max_connections, Reactor, CHANNEL_HEADROOM};
6use bytes::{BufMut, Bytes};
7use log::{debug, error, info, warn};
8use rml_rtmp::chunk_io::ChunkSerializer;
9use rml_rtmp::messages::{MessagePayload, RtmpMessage};
10use rml_rtmp::rml_amf0::Amf0Value;
11use rml_rtmp::time::RtmpTimestamp;
12use std::collections::HashMap;
13use std::marker::PhantomData;
14use std::net::{Shutdown, TcpListener, TcpStream};
15use std::sync::atomic::{AtomicUsize, Ordering};
16use std::sync::Arc;
17
18#[derive(Clone)]
19pub struct Initialization;
20#[derive(Clone)]
21pub struct Running;
22#[derive(Clone)]
23pub struct Ended;
24
25#[derive(Clone)]
26pub struct EmbedRtmpServer<S> {
27    address: String,
28    bound_addr: Option<std::net::SocketAddr>,
29    status: Arc<AtomicUsize>,
30    stream_keys: dashmap::DashSet<String>,
31    // stream_key bytes_receiver
32    publisher_sender: Option<crossbeam_channel::Sender<(String, crossbeam_channel::Receiver<Vec<u8>>)>>,
33    gop_limit: usize,
34    max_connections: Option<usize>,
35    state: PhantomData<S>,
36}
37
38const STATUS_INIT: usize = 0;
39const STATUS_RUN: usize = 1;
40const STATUS_END: usize = 2;
41
42impl<S: 'static> EmbedRtmpServer<S> {
43    fn into_state<T>(self) -> EmbedRtmpServer<T> {
44        EmbedRtmpServer {
45            address: self.address,
46            bound_addr: self.bound_addr,
47            status: self.status,
48            stream_keys: self.stream_keys,
49            publisher_sender: self.publisher_sender,
50            gop_limit: self.gop_limit,
51            max_connections: self.max_connections,
52            state: Default::default(),
53        }
54    }
55
56    /// Checks whether the RTMP server has been stopped. This returns `true` after
57    /// [`stop`](EmbedRtmpServer<Running>::stop) has been called and the server has exited its main loop, otherwise `false`.
58    ///
59    /// # Returns
60    ///
61    /// * `true` if the server has been signaled to stop (and is no longer listening/accepting).
62    /// * `false` if the server is still running.
63    pub fn is_stopped(&self) -> bool {
64        self.status.load(Ordering::Acquire) == STATUS_END
65    }
66}
67
68impl EmbedRtmpServer<Initialization> {
69    /// Creates a new RTMP server instance that will listen on the specified address
70    /// when [`start`](EmbedRtmpServer<Initialization>::start) is called.
71    ///
72    /// # Parameters
73    ///
74    /// * `address` - A string slice representing the address (host:port) to bind the
75    ///   RTMP server socket.
76    ///
77    /// # Returns
78    ///
79    /// An [`EmbedRtmpServer`] configured to listen on the given address.
80    pub fn new(address: impl Into<String>) -> EmbedRtmpServer<Initialization> {
81        Self::new_with_gop_limit(address, 1)
82    }
83
84    /// Creates a new RTMP server instance that will listen on the specified address,
85    /// with a custom GOP limit.
86    ///
87    /// This method allows specifying the maximum number of GOPs to be cached.
88    /// A GOP (Group of Pictures) represents a sequence of video frames (I, P, B frames)
89    /// used for efficient video decoding and random access. The GOP limit defines
90    /// how many such groups are stored in the cache.
91    ///
92    /// # Parameters
93    ///
94    /// * `address` - A string slice representing the address (host:port) to bind the
95    ///   RTMP server socket.
96    /// * `gop_limit` - The maximum number of GOPs to cache.
97    ///
98    /// # Returns
99    ///
100    /// An [`EmbedRtmpServer`] instance configured to listen on the given address and
101    /// using the specified GOP limit.
102    pub fn new_with_gop_limit(address: impl Into<String>, gop_limit: usize) -> EmbedRtmpServer<Initialization> {
103        Self {
104            address: address.into(),
105            bound_addr: None,
106            status: Arc::new(AtomicUsize::new(STATUS_INIT)),
107            stream_keys: Default::default(),
108            publisher_sender: None,
109            gop_limit,
110            max_connections: None,
111            state: Default::default(),
112        }
113    }
114
115    /// Sets the maximum number of concurrent connections allowed.
116    ///
117    /// If not set, the limit is auto-detected based on system file descriptor limits
118    /// (default: 10000, capped at 80% of system FD limit).
119    ///
120    /// # Parameters
121    ///
122    /// * `max_connections` - Maximum number of concurrent connections
123    ///
124    /// # Returns
125    ///
126    /// Self for method chaining.
127    pub fn set_max_connections(mut self, max_connections: usize) -> Self {
128        self.max_connections = Some(max_connections);
129        self
130    }
131
132    /// Starts the RTMP server on the configured address, entering a loop that
133    /// accepts incoming client connections. This method spawns background threads
134    /// to handle the connections and publish events.
135    ///
136    /// # Returns
137    ///
138    /// * `Ok(())` if the server successfully starts listening.
139    /// * An error variant if the socket could not be bound or other I/O errors occur.
140    pub fn start(mut self) -> crate::error::Result<EmbedRtmpServer<Running>> {
141        let listener = TcpListener::bind(self.address.clone())
142            .map_err(|e| <std::io::Error as Into<crate::error::Error>>::into(e))?;
143
144        // Get actual bound address (important for port 0)
145        let actual_addr = listener.local_addr()
146            .map_err(|e| <std::io::Error as Into<crate::error::Error>>::into(e))?;
147        self.bound_addr = Some(actual_addr);
148
149        listener
150            .set_nonblocking(true)
151            .map_err(|e| <std::io::Error as Into<crate::error::Error>>::into(e))?;
152
153        self.status.store(STATUS_RUN, Ordering::Release);
154
155        // Calculate effective max and create bounded channel with headroom
156        // This prevents unbounded queue growth when reactor is at capacity
157        let effective_max = effective_max_connections(self.max_connections);
158        let channel_capacity = effective_max.saturating_add(CHANNEL_HEADROOM);
159        let (stream_sender, stream_receiver) = crossbeam_channel::bounded(channel_capacity);
160        let (publisher_sender, publisher_receiver) = crossbeam_channel::bounded(1024);
161        self.publisher_sender = Some(publisher_sender);
162        let stream_keys = self.stream_keys.clone();
163        let status = self.status.clone();
164        let max_connections = self.max_connections;
165        let result = std::thread::Builder::new()
166            .name("rtmp-server-worker".to_string())
167            .spawn(move || handle_connections(stream_receiver, publisher_receiver, stream_keys, self.gop_limit, max_connections, status));
168        if let Err(e) = result {
169            error!("Thread[rtmp-server-worker] exited with error: {e}");
170            return Err(crate::error::Error::RtmpThreadExited);
171        }
172
173        info!(
174            "Embed rtmp server listening for connections on {} (actual: {}, max_connections: {}).",
175            &self.address, actual_addr, effective_max
176        );
177
178        let status = self.status.clone();
179        let result = std::thread::Builder::new()
180            .name("rtmp-server-io".to_string())
181            .spawn(move || {
182            for stream in listener.incoming() {
183                // Check the stop flag on every iteration, not only when the
184                // listener runs dry: under a steady stream of incoming
185                // connections the WouldBlock branch is never taken and
186                // stop() would otherwise never terminate this thread.
187                if status.load(Ordering::Acquire) == STATUS_END {
188                    info!("Embed rtmp server stopped.");
189                    break;
190                }
191                match stream {
192                    Ok(stream) => {
193                        // Use try_send to apply backpressure when channel is full
194                        match stream_sender.try_send(stream) {
195                            Ok(_) => {
196                                debug!("New rtmp connection accepted.");
197                            }
198                            Err(crossbeam_channel::TrySendError::Full(s)) => {
199                                // Channel full - server at capacity, reject connection immediately
200                                let _ = s.shutdown(Shutdown::Both);
201                                debug!("Connection rejected: server at capacity (channel full)");
202                            }
203                            Err(crossbeam_channel::TrySendError::Disconnected(_)) => {
204                                error!("Connection channel disconnected");
205                                status.store(STATUS_END, Ordering::Release);
206                                return;
207                            }
208                        }
209                    }
210                    Err(e) => {
211                        if e.kind() == std::io::ErrorKind::WouldBlock {
212                            std::thread::sleep(std::time::Duration::from_millis(100));
213                        } else if is_fd_exhaustion(&e) {
214                            // Accepting again immediately would fail the same
215                            // way and spin the CPU; back off and let existing
216                            // connections close first.
217                            warn!("Accept failed, file descriptors exhausted: {e}");
218                            std::thread::sleep(std::time::Duration::from_millis(100));
219                        } else {
220                            debug!("Rtmp connection error: {:?}", e);
221                        }
222                    }
223                }
224            }
225        });
226        if let Err(e) = result {
227            error!("Thread[rtmp-server-io] exited with error: {e}");
228            return Err(crate::error::Error::RtmpThreadExited);
229        }
230
231        Ok(self.into_state())
232    }
233}
234
235impl EmbedRtmpServer<Running> {
236    /// Returns the actual bound socket address of the RTMP server.
237    ///
238    /// This is particularly useful when binding to port 0 (random port allocation),
239    /// as it allows you to discover which port the OS assigned.
240    ///
241    /// # Returns
242    ///
243    /// * `Option<std::net::SocketAddr>` - The actual bound address, or `None` if not available.
244    ///
245    /// # Example
246    ///
247    /// ```rust,ignore
248    /// let server = EmbedRtmpServer::new("127.0.0.1:0").start().unwrap();
249    /// let actual_port = server.local_addr().unwrap().port();
250    /// println!("Server listening on port: {}", actual_port);
251    /// ```
252    pub fn local_addr(&self) -> Option<std::net::SocketAddr> {
253        self.bound_addr
254    }
255
256    /// Creates an RTMP "input" endpoint for this server (from the server's perspective),
257    /// returning an [`Output`] that can be used by FFmpeg to push media data.
258    ///
259    /// From the FFmpeg standpoint, the returned [`Output`] is where media content is
260    /// sent (i.e., FFmpeg "outputs" to this RTMP server). After obtaining this [`Output`],
261    /// you can pass it to your FFmpeg job or scheduler to start streaming data into the server.
262    ///
263    /// # Parameters
264    ///
265    /// * `app_name` - The RTMP application name, typically corresponding to the `app` part
266    ///   of an RTMP URL (e.g., `rtmp://host:port/app/stream_key`).
267    /// * `stream_key` - The stream key (or "stream name"). If a stream with the same key
268    ///   already exists, an error will be returned.
269    ///
270    /// # Returns
271    ///
272    /// * [`Output`] - An output object preconfigured for streaming to this RTMP server.
273    ///   This can be passed to the FFmpeg SDK for actual data push.
274    /// * [`crate::error::Error`] - If a stream with the same key already exists, the server
275    ///   is not ready, or an internal error occurs, the corresponding error is returned.
276    ///
277    /// # Example
278    ///
279    /// ```rust,ignore
280    /// # // Assume there are definitions and initializations for FfmpegContext, FfmpegScheduler, etc.
281    ///
282    /// // 1. Create and start the RTMP server
283    /// let mut rtmp_server = EmbedRtmpServer::new("localhost:1935");
284    /// rtmp_server.start().expect("Failed to start RTMP server");
285    ///
286    /// // 2. Create an RTMP "input" with app_name="my-app" and stream_key="my-stream"
287    /// let output = rtmp_server
288    ///     .create_rtmp_input("my-app", "my-stream")
289    ///     .expect("Failed to create RTMP input");
290    ///
291    /// // 3. Prepare the FFmpeg context to push a local file to the newly created `Output`
292    /// let context = FfmpegContext::builder()
293    ///     .input("test.mp4")
294    ///     .output(output)
295    ///     .build()
296    ///     .expect("Failed to build Ffmpeg context");
297    ///
298    /// // 4. Start FFmpeg to push "test.mp4" to the local RTMP server on "my-app/my-stream"
299    /// FfmpegScheduler::new(context)
300    ///     .start()
301    ///     .expect("Failed to start Ffmpeg job");
302    /// ```
303    pub fn create_rtmp_input(
304        &self,
305        app_name: impl Into<String>,
306        stream_key: impl Into<String>,
307    ) -> crate::error::Result<Output> {
308        let message_sender = self.create_stream_sender(app_name, stream_key)?;
309
310        let mut flv_buffer = FlvBuffer::new();
311        let mut serializer = ChunkSerializer::new();
312        let write_callback: Box<dyn FnMut(&[u8]) -> i32 + Send> = Box::new(move |buf: &[u8]| -> i32 {
313            flv_buffer.write_data(buf);
314            // One AVIO write can carry many FLV tags (the muxer hands over
315            // 64KB blocks): drain every complete tag now, or the backlog
316            // grows and the final tags of the stream are never sent.
317            while let Some(mut flv_tag) = flv_buffer.get_flv_tag() {
318                flv_tag.header.stream_id = 1;
319                match serializer.serialize(&flv_tag_to_message_payload(flv_tag), false, true) {
320                    Ok(packet) => {
321                        if let Err(e) = message_sender.send(packet.bytes) {
322                            error!("Failed to send RTMP packet: {:?}", e);
323                            return -1;
324                        }
325                    }
326                    Err(e) => {
327                        error!("Failed to serialize RTMP message: {:?}", e);
328                        return -1;
329                    }
330                }
331            }
332            buf.len() as i32
333        });
334
335        let output: Output = write_callback.into();
336
337        Ok(output
338            .set_format("flv")
339            .set_video_codec("h264")
340            .set_audio_codec("aac")
341            .set_format_opt("flvflags", "no_duration_filesize"))
342    }
343
344    /// Creates a sender channel for an RTMP stream, identified by `app_name` and `stream_key`.
345    /// This method is used internally by [`create_rtmp_input`](EmbedRtmpServer<Running>::create_rtmp_input) but can also be called directly
346    /// if you need more control over how the stream is handled.
347    ///
348    /// # Parameters
349    ///
350    /// * `app_name` - The RTMP application name.
351    /// * `stream_key` - The unique name (or key) for this stream. Must not already be in use.
352    ///
353    /// # Returns
354    ///
355    /// * `crossbeam_channel::Sender<Vec<u8>>` - A sender that allows you to send raw RTMP bytes
356    ///   into the server's handling pipeline.
357    /// * [`crate::error::Error`] - If a stream with the same key already exists or other
358    ///   internal issues occur, an error is returned.
359    ///
360    /// # Notes
361    ///
362    /// * This function sets up the initial RTMP "connect" and "publish" commands automatically.
363    /// * If you manually send bytes to the resulting channel, they should already be properly
364    ///   packaged as RTMP chunks. Otherwise, the server might fail to parse them.
365    pub fn create_stream_sender(
366        &self,
367        app_name: impl Into<String>,
368        stream_key: impl Into<String>,
369    ) -> crate::error::Result<crossbeam_channel::Sender<Vec<u8>>> {
370        let stream_key = stream_key.into();
371        if self.stream_keys.contains(&stream_key) {
372            return Err(RtmpStreamAlreadyExists(stream_key));
373        }
374
375        let (sender, receiver) = crossbeam_channel::bounded(1024);
376
377        let publisher_sender = match self.publisher_sender.as_ref() {
378            Some(sender) => sender,
379            None => {
380                error!("Publisher sender not initialized");
381                return Err(RtmpCreateStream.into());
382            }
383        };
384
385        if let Err(_) = publisher_sender.send((stream_key.clone(), receiver)) {
386            if self.status.load(Ordering::Acquire) != STATUS_END {
387                warn!("Rtmp server worker already exited. Can't create stream sender.");
388            } else {
389                error!("Rtmp Server aborted. Can't create stream sender.");
390            }
391            return Err(RtmpCreateStream.into());
392        }
393
394        let mut serializer = ChunkSerializer::new();
395
396        // send connect
397        let mut properties: HashMap<String, Amf0Value> = HashMap::new();
398        properties.insert("app".to_string(), Amf0Value::Utf8String(app_name.into()));
399        let connect_cmd = RtmpMessage::Amf0Command {
400            command_name: "connect".to_string(),
401            transaction_id: 1.0,
402            command_object: Amf0Value::Object(properties),
403            additional_arguments: Vec::new(),
404        }
405        .into_message_payload(RtmpTimestamp { value: 0 }, 0);
406
407        let connect_cmd = match connect_cmd {
408            Ok(cmd) => cmd,
409            Err(e) => {
410                error!("Failed to create connect command: {:?}", e);
411                return Err(RtmpCreateStream.into());
412            }
413        };
414
415        let connect_packet = match serializer.serialize(&connect_cmd, false, true) {
416            Ok(packet) => packet,
417            Err(e) => {
418                error!("Failed to serialize connect command: {:?}", e);
419                return Err(RtmpCreateStream.into());
420            }
421        };
422
423        if let Err(_) = sender.send(connect_packet.bytes) {
424            error!("Can't send connect command to rtmp server.");
425            return Err(RtmpCreateStream.into());
426        }
427
428        // send createStream
429        let create_stream_cmd = RtmpMessage::Amf0Command {
430            command_name: "createStream".to_string(),
431            transaction_id: 2.0,
432            command_object: Amf0Value::Null,
433            additional_arguments: Vec::new(),
434        }
435        .into_message_payload(RtmpTimestamp { value: 0 }, 1);
436
437        let create_stream_cmd = match create_stream_cmd {
438            Ok(cmd) => cmd,
439            Err(e) => {
440                error!("Failed to create createStream command: {:?}", e);
441                return Err(RtmpCreateStream.into());
442            }
443        };
444
445        let create_stream_packet = match serializer.serialize(&create_stream_cmd, false, true) {
446            Ok(packet) => packet,
447            Err(e) => {
448                error!("Failed to serialize createStream command: {:?}", e);
449                return Err(RtmpCreateStream.into());
450            }
451        };
452
453        if let Err(_) = sender.send(create_stream_packet.bytes) {
454            error!("Can't send createStream command to rtmp server.");
455            return Err(RtmpCreateStream.into());
456        }
457
458        // send publish
459        let mut arguments = Vec::new();
460        arguments.push(Amf0Value::Utf8String(stream_key));
461        arguments.push(Amf0Value::Utf8String("live".into()));
462        let publish_cmd = RtmpMessage::Amf0Command {
463            command_name: "publish".to_string(),
464            transaction_id: 3.0,
465            command_object: Amf0Value::Null,
466            additional_arguments: arguments,
467        }
468        .into_message_payload(RtmpTimestamp { value: 0 }, 1);
469
470        let publish_cmd = match publish_cmd {
471            Ok(cmd) => cmd,
472            Err(e) => {
473                error!("Failed to create publish command: {:?}", e);
474                return Err(RtmpCreateStream.into());
475            }
476        };
477
478        let publish_packet = match serializer.serialize(&publish_cmd, false, true) {
479            Ok(packet) => packet,
480            Err(e) => {
481                error!("Failed to serialize publish command: {:?}", e);
482                return Err(RtmpCreateStream.into());
483            }
484        };
485
486        if let Err(_) = sender.send(publish_packet.bytes) {
487            error!("Can't send publish command to rtmp server.");
488            return Err(RtmpCreateStream.into());
489        }
490        Ok(sender)
491    }
492
493    /// Stops the RTMP server by signaling the listening and connection-handling threads
494    /// to terminate. Once called, new incoming connections will be ignored, and existing
495    /// threads will exit gracefully.
496    ///
497    /// # Example
498    /// ```rust,ignore
499    /// let server = EmbedRtmpServer::new("localhost:1935");
500    /// // ... start and handle streaming
501    /// server.stop();
502    /// assert!(server.is_stopped());
503    /// ```
504    pub fn stop(self) -> EmbedRtmpServer<Ended> {
505        self.status.store(STATUS_END, Ordering::Release);
506        self.into_state()
507    }
508}
509
510/// Handle connections using optimized Reactor
511///
512/// Replaces old multi-threaded handle_connections with single-threaded event-driven model:
513/// - Uses epoll/kqueue/WSAPoll for IO multiplexing
514/// - Write queue with backpressure management
515/// - Strict drain until WouldBlock semantics
516/// Whether an accept error means the process (EMFILE) or system (ENFILE) ran
517/// out of file descriptors. `io::ErrorKind` has no stable variant for these,
518/// so match on the raw OS error code.
519fn is_fd_exhaustion(e: &std::io::Error) -> bool {
520    #[cfg(unix)]
521    {
522        matches!(e.raw_os_error(), Some(code) if code == libc::EMFILE || code == libc::ENFILE)
523    }
524    #[cfg(windows)]
525    {
526        // WSAEMFILE: too many open sockets.
527        e.raw_os_error() == Some(10024)
528    }
529    #[cfg(not(any(unix, windows)))]
530    {
531        let _ = e;
532        false
533    }
534}
535
536fn handle_connections(
537    connection_receiver: crossbeam_channel::Receiver<TcpStream>,
538    publisher_receiver: crossbeam_channel::Receiver<(String, crossbeam_channel::Receiver<Vec<u8>>)>,
539    stream_keys: dashmap::DashSet<String>,
540    gop_limit: usize,
541    max_connections: Option<usize>,
542    status: Arc<AtomicUsize>,
543) {
544    // Create Reactor
545    let mut reactor = match Reactor::new(gop_limit, max_connections, stream_keys, status.clone()) {
546        Ok(r) => r,
547        Err(e) => {
548            error!("Failed to create Reactor: {:?}", e);
549            status.store(STATUS_END, Ordering::Release);
550            return;
551        }
552    };
553
554    // Run Reactor main loop
555    reactor.run(connection_receiver, publisher_receiver);
556
557    if status.load(Ordering::Acquire) != STATUS_END {
558        error!("Rtmp Server aborted.");
559    }
560}
561
562pub fn flv_tag_to_message_payload(flv_tag: FlvTag) -> MessagePayload {
563    let timestamp = flv_tag.header.timestamp | ((flv_tag.header.timestamp_ext as u32) << 24);
564
565    let type_id = flv_tag.header.tag_type;
566    let message_stream_id = flv_tag.header.stream_id;
567
568    let data = if type_id == 0x12 {
569        wrap_metadata(flv_tag.data)
570    } else {
571        flv_tag.data
572    };
573
574    MessagePayload {
575        timestamp: RtmpTimestamp { value: timestamp },
576        type_id,
577        message_stream_id,
578        data,
579    }
580}
581
582fn wrap_metadata(data: Bytes) -> Bytes {
583    let s = "@setDataFrame";
584
585    let insert_len = 16;
586
587    let mut bytes = bytes::BytesMut::with_capacity(insert_len + data.len());
588
589    bytes.put_u8(0x02);
590    bytes.put_u16(s.len() as u16);
591    bytes.put(s.as_bytes());
592
593    bytes.put(data);
594
595    bytes.freeze()
596}
597
598
599// ============================================================================
600// StreamBuilder API - Simplified RTMP streaming interface
601// ============================================================================
602
603use crate::core::context::ffmpeg_context::FfmpegContext;
604use crate::core::context::input::Input;
605use crate::core::scheduler::ffmpeg_scheduler::{FfmpegScheduler, Running as SchedulerRunning};
606use crate::error::StreamError;
607use std::path::{Path, PathBuf};
608
609/// A builder for creating RTMP streaming sessions with a simplified API.
610///
611/// This builder provides a fluent interface for configuring and starting
612/// RTMP streaming without needing to manually manage the server lifecycle.
613///
614/// # Example
615///
616/// ```rust,ignore
617/// use ez_ffmpeg::rtmp::embed_rtmp_server::EmbedRtmpServer;
618///
619/// let handle = EmbedRtmpServer::stream_builder()
620///     .address("localhost:1935")
621///     .app_name("live")
622///     .stream_key("stream1")
623///     .input_file("video.mp4")
624///     // readrate defaults to 1.0 (realtime)
625///     .start()?;
626///
627/// handle.wait()?;
628/// ```
629pub struct StreamBuilder {
630    address: Option<String>,
631    app_name: Option<String>,
632    stream_key: Option<String>,
633    input_file: Option<PathBuf>,
634    readrate: Option<f32>,
635    gop_limit: Option<usize>,
636    max_connections: Option<usize>,
637}
638
639impl Default for StreamBuilder {
640    fn default() -> Self {
641        Self::new()
642    }
643}
644
645impl StreamBuilder {
646    /// Creates a new `StreamBuilder` with default settings.
647    ///
648    /// By default, `readrate` is set to `1.0` (real-time playback speed),
649    /// which is equivalent to FFmpeg's `-re` flag. This is the recommended
650    /// setting for live RTMP streaming scenarios.
651    pub fn new() -> Self {
652        Self {
653            address: None,
654            app_name: None,
655            stream_key: None,
656            input_file: None,
657            readrate: Some(1.0), // Default to real-time speed for live streaming
658            gop_limit: None,
659            max_connections: None,
660        }
661    }
662
663    /// Sets the address for the RTMP server (e.g., "localhost:1935").
664    pub fn address(mut self, address: impl Into<String>) -> Self {
665        self.address = Some(address.into());
666        self
667    }
668
669    /// Sets the RTMP application name.
670    pub fn app_name(mut self, app_name: impl Into<String>) -> Self {
671        self.app_name = Some(app_name.into());
672        self
673    }
674
675    /// Sets the stream key (publishing name).
676    pub fn stream_key(mut self, stream_key: impl Into<String>) -> Self {
677        self.stream_key = Some(stream_key.into());
678        self
679    }
680
681    /// Sets the input file path to stream.
682    pub fn input_file(mut self, path: impl AsRef<Path>) -> Self {
683        self.input_file = Some(path.as_ref().to_path_buf());
684        self
685    }
686
687    /// Sets the read rate for the input file.
688    ///
689    /// A value of 1.0 means realtime playback speed.
690    /// This is useful for simulating live streaming from a file.
691    pub fn readrate(mut self, rate: f32) -> Self {
692        self.readrate = Some(rate);
693        self
694    }
695
696    /// Sets the GOP (Group of Pictures) limit for the RTMP server.
697    ///
698    /// This controls how many GOPs are buffered for new subscribers.
699    pub fn gop_limit(mut self, limit: usize) -> Self {
700        self.gop_limit = Some(limit);
701        self
702    }
703
704    /// Sets the maximum number of connections the server will accept.
705    pub fn max_connections(mut self, max: usize) -> Self {
706        self.max_connections = Some(max);
707        self
708    }
709
710    /// Starts the RTMP streaming session.
711    ///
712    /// This method validates all required parameters, starts the RTMP server,
713    /// and begins streaming the input file.
714    ///
715    /// # Required Parameters
716    ///
717    /// - `address`: The server address
718    /// - `app_name`: The RTMP application name
719    /// - `stream_key`: The stream key (publishing name)
720    /// - `input_file`: The file to stream
721    ///
722    /// # Returns
723    ///
724    /// A `StreamHandle` that can be used to wait for completion or manage the stream.
725    ///
726    /// # Errors
727    ///
728    /// Returns `StreamError` if:
729    /// - Any required parameter is missing
730    /// - The input file does not exist
731    /// - The server fails to start
732    /// - FFmpeg context creation fails
733    pub fn start(self) -> Result<StreamHandle, StreamError> {
734        // Validate required parameters
735        let address = self
736            .address
737            .ok_or(StreamError::MissingParameter("address"))?;
738        let app_name = self
739            .app_name
740            .ok_or(StreamError::MissingParameter("app_name"))?;
741        let stream_key = self
742            .stream_key
743            .ok_or(StreamError::MissingParameter("stream_key"))?;
744        let input_file = self
745            .input_file
746            .ok_or(StreamError::MissingParameter("input_file"))?;
747
748        // Validate input file exists and is a file (not a directory)
749        if !input_file.is_file() {
750            return Err(StreamError::InputNotFound { path: input_file });
751        }
752
753        // Create and configure the server
754        let mut server = if let Some(gop_limit) = self.gop_limit {
755            EmbedRtmpServer::new_with_gop_limit(&address, gop_limit)
756        } else {
757            EmbedRtmpServer::new(&address)
758        };
759
760        if let Some(max_conn) = self.max_connections {
761            server = server.set_max_connections(max_conn);
762        }
763
764        // Start the server
765        let server = server.start().map_err(StreamError::Ffmpeg)?;
766        let server = Arc::new(server);
767
768        // Create the RTMP output
769        let output = server
770            .create_rtmp_input(&app_name, &stream_key)
771            .map_err(StreamError::Ffmpeg)?;
772
773        // Create the input with optional readrate
774        let input_path = input_file.to_string_lossy().to_string();
775        let mut input = Input::from(input_path);
776        if let Some(rate) = self.readrate {
777            input = input.set_readrate(rate);
778        }
779
780        // Build and start the FFmpeg context
781        let scheduler = FfmpegContext::builder()
782            .input(input)
783            .output(output)
784            .build()
785            .map_err(StreamError::Ffmpeg)?
786            .start()
787            .map_err(StreamError::Ffmpeg)?;
788
789        Ok(StreamHandle {
790            _server: server,
791            scheduler: Some(scheduler),
792        })
793    }
794}
795
796/// A handle to a running RTMP streaming session.
797///
798/// This handle manages the lifecycle of both the RTMP server and the FFmpeg
799/// streaming context. When dropped, it will attempt to clean up resources.
800///
801/// # Example
802///
803/// ```rust,ignore
804/// let handle = EmbedRtmpServer::stream_builder()
805///     .address("localhost:1935")
806///     .app_name("live")
807///     .stream_key("stream1")
808///     .input_file("video.mp4")
809///     .start()?;
810///
811/// // Wait for streaming to complete
812/// handle.wait()?;
813/// ```
814pub struct StreamHandle {
815    _server: Arc<EmbedRtmpServer<Running>>,
816    scheduler: Option<FfmpegScheduler<SchedulerRunning>>,
817}
818
819impl StreamHandle {
820    /// Waits for the streaming session to complete.
821    ///
822    /// This method blocks until the FFmpeg context finishes processing
823    /// (e.g., when the input file ends or an error occurs).
824    ///
825    /// # Returns
826    ///
827    /// Returns `Ok(())` if streaming completed successfully, or an error
828    /// if something went wrong during streaming.
829    pub fn wait(mut self) -> Result<(), StreamError> {
830        if let Some(scheduler) = self.scheduler.take() {
831            scheduler.wait().map_err(StreamError::Ffmpeg)?;
832        }
833        Ok(())
834    }
835}
836
837impl Drop for StreamHandle {
838    fn drop(&mut self) {
839        // Best-effort cleanup: if scheduler wasn't consumed by wait(),
840        // we attempt to stop it gracefully here.
841        // The server will be stopped when the Arc is dropped.
842        if let Some(scheduler) = self.scheduler.take() {
843            // Attempt to wait for graceful shutdown, but don't block forever
844            let _ = scheduler.wait();
845        }
846    }
847}
848
849impl EmbedRtmpServer<Initialization> {
850    /// Creates a new `StreamBuilder` for simplified RTMP streaming.
851    ///
852    /// This is the recommended entry point for simple streaming scenarios
853    /// where you want to stream a file to an embedded RTMP server.
854    ///
855    /// # Example
856    ///
857    /// ```rust,ignore
858    /// use ez_ffmpeg::rtmp::embed_rtmp_server::EmbedRtmpServer;
859    ///
860    /// let handle = EmbedRtmpServer::stream_builder()
861    ///     .address("localhost:1935")
862    ///     .app_name("live")
863    ///     .stream_key("stream1")
864    ///     .input_file("video.mp4")
865    ///     .start()?;
866    ///
867    /// handle.wait()?;
868    /// ```
869    ///
870    /// For more complex scenarios requiring full control over the server
871    /// and FFmpeg context, use the traditional API:
872    ///
873    /// ```rust,ignore
874    /// let server = EmbedRtmpServer::new("localhost:1935").start()?;
875    /// let output = server.create_rtmp_input("app", "stream")?;
876    /// // ... configure Input and FfmpegContext manually
877    /// ```
878    pub fn stream_builder() -> StreamBuilder {
879        StreamBuilder::new()
880    }
881}
882
883#[cfg(test)]
884mod tests {
885    use super::*;
886    use crate::core::context::ffmpeg_context::FfmpegContext;
887    use crate::core::context::input::Input;
888    use crate::core::context::output::Output;
889    use crate::core::scheduler::ffmpeg_scheduler::FfmpegScheduler;
890    use ffmpeg_next::time::current;
891    use std::thread::sleep;
892    use std::time::Duration;
893
894    #[test]
895    #[ignore] // Integration test: requires exclusive port 1935 and test.mp4
896    fn test_concat_stream_loop() {
897        let _ = env_logger::builder()
898            .filter_level(log::LevelFilter::Trace)
899            .is_test(true)
900            .try_init();
901
902        let embed_rtmp_server = EmbedRtmpServer::new("localhost:1935");
903        let embed_rtmp_server = embed_rtmp_server.start().unwrap();
904
905        let output = embed_rtmp_server
906            .create_rtmp_input("my-app", "my-stream")
907            .unwrap();
908
909        let start = current();
910
911        let result = FfmpegContext::builder()
912            .input(Input::from("test.mp4")
913                .set_readrate(1.0)
914                .set_stream_loop(3)
915            )
916            .input(
917                Input::from("test.mp4")
918                    .set_readrate(1.0)
919                    .set_stream_loop(3)
920            )
921            .input(
922                Input::from("test.mp4")
923                    .set_readrate(1.0)
924                    .set_stream_loop(3)
925            )
926            .filter_desc("[0:v][0:a][1:v][1:a][2:v][2:a]concat=n=3:v=1:a=1")
927            .output(output)
928            .build()
929            .unwrap()
930            .start()
931            .unwrap()
932            .wait();
933
934        assert!(result.is_ok());
935        info!("elapsed time: {}", current() - start);
936    }
937
938    #[test]
939    #[ignore] // Integration test: requires exclusive port 1935 and test.mp4
940    fn test_stream_loop() {
941        let _ = env_logger::builder()
942            .filter_level(log::LevelFilter::Trace)
943            .is_test(true)
944            .try_init();
945
946        let embed_rtmp_server = EmbedRtmpServer::new("localhost:1935");
947        let embed_rtmp_server = embed_rtmp_server.start().unwrap();
948
949        let output = embed_rtmp_server
950            .create_rtmp_input("my-app", "my-stream")
951            .unwrap();
952
953        let start = current();
954
955        let result = FfmpegContext::builder()
956            .input(Input::from("test.mp4").set_readrate(1.0).set_stream_loop(-1))
957            // .filter_desc("hue=s=0")
958            .output(output.set_video_codec("h264_videotoolbox"))
959            .build()
960            .unwrap()
961            .start()
962            .unwrap()
963            .wait();
964
965        assert!(result.is_ok());
966
967        info!("elapsed time: {}", current() - start);
968    }
969
970    #[test]
971    #[ignore] // Integration test: requires exclusive port 1935 and test.mp4
972    fn test_concat_realtime() {
973        let _ = env_logger::builder()
974            .filter_level(log::LevelFilter::Trace)
975            .is_test(true)
976            .try_init();
977
978        let embed_rtmp_server = EmbedRtmpServer::new("localhost:1935");
979        let embed_rtmp_server = embed_rtmp_server.start().unwrap();
980
981        let output = embed_rtmp_server
982            .create_rtmp_input("my-app", "my-stream")
983            .unwrap();
984
985        let start = current();
986
987        let result = FfmpegContext::builder()
988            .independent_readrate()
989            .input(Input::from("test.mp4").set_readrate(1.0))
990            .input(
991                Input::from("test.mp4")
992                    .set_readrate(1.0)
993            )
994            .input(
995                Input::from("test.mp4")
996                    .set_readrate(1.0)
997            )
998            .filter_desc("[0:v][0:a][1:v][1:a][2:v][2:a]concat=n=3:v=1:a=1")
999            .output(output)
1000            .build()
1001            .unwrap()
1002            .start()
1003            .unwrap()
1004            .wait();
1005
1006        assert!(result.is_ok());
1007
1008        sleep(Duration::from_secs(1));
1009        info!("elapsed time: {}", current() - start);
1010    }
1011
1012    #[test]
1013    #[ignore] // Integration test: requires exclusive port 1935 and test.mp4
1014    fn test_realtime() {
1015        let _ = env_logger::builder()
1016            .filter_level(log::LevelFilter::Trace)
1017            .is_test(true)
1018            .try_init();
1019
1020        let embed_rtmp_server = EmbedRtmpServer::new("localhost:1935");
1021        let embed_rtmp_server = embed_rtmp_server.start().unwrap();
1022
1023        let output = embed_rtmp_server
1024            .create_rtmp_input("my-app", "my-stream")
1025            .unwrap();
1026
1027        let start = current();
1028
1029        let result = FfmpegContext::builder()
1030            .input(Input::from("test.mp4").set_readrate(1.0))
1031            .output(output)
1032            .build()
1033            .unwrap()
1034            .start()
1035            .unwrap()
1036            .wait();
1037
1038        assert!(result.is_ok());
1039
1040        info!("elapsed time: {}", current() - start);
1041    }
1042
1043    #[test]
1044    #[ignore] // Integration test: requires test.mp4
1045    fn test_readrate() {
1046        let _ = env_logger::builder()
1047            .filter_level(log::LevelFilter::Trace)
1048            .is_test(true)
1049            .try_init();
1050
1051        let mut output: Output = "output.flv".into();
1052        output.audio_codec = Some("adpcm_swf".to_string());
1053
1054        let mut input: Input = "test.mp4".into();
1055        input.readrate = Some(1.0);
1056
1057        let context = FfmpegContext::builder()
1058            .input(input)
1059            .output(output)
1060            .build()
1061            .unwrap();
1062
1063        let result = FfmpegScheduler::new(context).start().unwrap().wait();
1064        if let Err(error) = result {
1065            println!("Error: {error}");
1066        }
1067    }
1068
1069    #[test]
1070    #[ignore] // Integration test: requires exclusive port 1935 and test.mp4
1071    fn test_embed_rtmp_server() {
1072        let _ = env_logger::builder()
1073            .filter_level(log::LevelFilter::Trace)
1074            .is_test(true)
1075            .try_init();
1076
1077        let embed_rtmp_server = EmbedRtmpServer::new("localhost:1935");
1078        let embed_rtmp_server = embed_rtmp_server.start().unwrap();
1079
1080        let output = embed_rtmp_server
1081            .create_rtmp_input("my-app", "my-stream")
1082            .unwrap();
1083        let mut input: Input = "test.mp4".into();
1084        input.readrate = Some(1.0);
1085
1086        let context = FfmpegContext::builder()
1087            .input(input)
1088            .output(output)
1089            .build()
1090            .unwrap();
1091
1092        let result = FfmpegScheduler::new(context).start().unwrap().wait();
1093
1094        assert!(result.is_ok());
1095
1096        sleep(Duration::from_secs(3));
1097    }
1098}
ez_ffmpeg/rtmp/embed_rtmp_server.rs

ez_ffmpeg/rtmp/
embed_rtmp_server.rs
