Struct EmbedRtmpServer

Source

pub struct EmbedRtmpServer<S> { /* private fields */ }

Implementations§

Source §

impl<S: 'static> EmbedRtmpServer<S>

Source

pub fn is_stopped(&self) -> bool

Checks whether the RTMP server has been signaled to stop. This returns true once stop has been called (or a fatal internal error stopped the server), otherwise false.

Note this reports the signal, not thread teardown: the worker threads observe the flag and exit shortly after (the reactor on its next wakeup, the accept thread within its ~100ms accept cycle). The one place both coincide is stop, which joins the server threads before returning — after a stop() call has returned, teardown is complete as well (except for stop()’s reentrant logger edge, documented there, which is signal-only).

§Returns

true if the server has been signaled to stop (and will no longer accept connections).
false if the server is still running.

Source §

impl EmbedRtmpServer<Initialization>

Source

pub fn new(address: impl Into<String>) -> EmbedRtmpServer<Initialization>

Creates a new RTMP server instance that will listen on the specified address when start is called.

§Parameters

address - A string slice representing the address (host:port) to bind the RTMP server socket.

§Returns

An EmbedRtmpServer configured to listen on the given address.

Source

pub fn new_with_gop_limit( address: impl Into<String>, gop_limit: usize, ) -> EmbedRtmpServer<Initialization>

Creates a new RTMP server instance that will listen on the specified address, with a custom GOP limit.

This method allows specifying the maximum number of GOPs to be cached. A GOP (Group of Pictures) represents a sequence of video frames (I, P, B frames) used for efficient video decoding and random access. The GOP limit defines how many such groups are stored in the cache.

§Parameters

address - A string slice representing the address (host:port) to bind the RTMP server socket.
gop_limit - The maximum number of GOPs to cache.

§Returns

An EmbedRtmpServer instance configured to listen on the given address and using the specified GOP limit.

Source

pub fn set_max_connections(self, max_connections: usize) -> Self

Sets the maximum number of concurrent connections allowed.

If not set, the limit is auto-detected based on system file descriptor limits (default: 10000, capped at 80% of system FD limit).

§Parameters

max_connections - Maximum number of concurrent connections

§Returns

Self for method chaining.

Source

pub fn start(self) -> Result<EmbedRtmpServer<Running>>

Starts the RTMP server on the configured address, entering a loop that accepts incoming client connections. This method spawns background threads to handle the connections and publish events.

Clones of one server value share a single lifecycle, and that lifecycle can be started only once: the first start() wins, and every later attempt — a sibling clone while the server runs, or any clone after it stopped — fails with RtmpServerAlreadyStarted, refused before any socket is bound: a second start on a fixed port reports the lifecycle error, not AddrInUse from the winner’s port. A start() that fails before it claims the lifecycle (e.g. the bind) leaves it untouched, so a clone may retry; from the claim on, a failing start() stops the lifecycle for good.

§Returns

Ok(()) if the server successfully starts listening.
An error variant if the socket could not be bound, this server value’s lifecycle was already started once, or other I/O errors occur.

Source §

impl EmbedRtmpServer<Running>

Source

pub fn local_addr(&self) -> Option<SocketAddr>

Returns the actual bound socket address of the RTMP server.

This is particularly useful when binding to port 0 (random port allocation), as it allows you to discover which port the OS assigned.

§Returns

Option<std::net::SocketAddr> - The actual bound address, or None if not available.

§Example

let server = EmbedRtmpServer::new("127.0.0.1:0").start().unwrap();
let actual_port = server.local_addr().unwrap().port();
println!("Server listening on port: {}", actual_port);

Source

pub fn create_rtmp_input( &self, app_name: impl Into<String>, stream_key: impl Into<String>, ) -> Result<Output>

Creates an RTMP “input” endpoint for this server (from the server’s perspective), returning an Output that can be used by FFmpeg to push media data.

From the FFmpeg standpoint, the returned Output is where media content is sent (i.e., FFmpeg “outputs” to this RTMP server). After obtaining this Output, you can pass it to your FFmpeg job or scheduler to start streaming data into the server.

§Parameters

app_name - The RTMP application name, typically corresponding to the app part of an RTMP URL (e.g., rtmp://host:port/app/stream_key).
stream_key - The stream key (or “stream name”). If a stream with the same key already exists, an error will be returned.

§Returns

Output - An output object preconfigured for streaming to this RTMP server. This can be passed to the FFmpeg SDK for actual data push.
crate::error::Error - If a stream with the same key already exists, the server is not ready, or an internal error occurs, the corresponding error is returned.

§Example


// 1. Create and start the RTMP server
let mut rtmp_server = EmbedRtmpServer::new("localhost:1935");
rtmp_server.start().expect("Failed to start RTMP server");

// 2. Create an RTMP "input" with app_name="my-app" and stream_key="my-stream"
let output = rtmp_server
    .create_rtmp_input("my-app", "my-stream")
    .expect("Failed to create RTMP input");

// 3. Prepare the FFmpeg context to push a local file to the newly created `Output`
let context = FfmpegContext::builder()
    .input("test.mp4")
    .output(output)
    .build()
    .expect("Failed to build Ffmpeg context");

// 4. Start FFmpeg to push "test.mp4" to the local RTMP server on "my-app/my-stream"
FfmpegScheduler::new(context)
    .start()
    .expect("Failed to start Ffmpeg job");

Source

pub fn create_stream_sender( &self, app_name: impl Into<String>, stream_key: impl Into<String>, ) -> Result<RtmpStreamSender>

Creates a sender channel for an RTMP stream, identified by app_name and stream_key. Call this to publish an in-process stream directly by feeding raw, pre-packaged RTMP chunk bytes into the server’s handling pipeline.

§Parameters

app_name - The RTMP application name.
stream_key - The unique name (or key) for this stream. Must not already be in use.

§Returns

RtmpStreamSender - A handle whose send feeds raw RTMP bytes into the server’s handling pipeline.
crate::error::Error - If a stream with the same key already exists or other internal issues occur, an error is returned.

§Notes

This function sets up the initial RTMP “connect” and “publish” commands automatically.
If you manually send bytes to the resulting channel, they should already be properly packaged as RTMP chunks. Otherwise, the server might fail to parse them.

Source

pub fn stop(self) -> EmbedRtmpServer<Ended>

Stops the RTMP server: signals the listening and connection-handling threads to terminate, then joins them, so this returns only once teardown has settled — the listener socket is released, every connection is closed, and every publisher is torn down with its stream key freed. A caller can rebind the address or reuse a stream key immediately after this returns, without polling.

Blocking is bounded but not hard-real-time: the reactor finishes the loop turn it is in, then drains still-queued tails for roughly its graceful-shutdown window (a few seconds, re-checked between drain passes, and only spent when a peer stopped reading); the accept thread notices within its ~100ms accept cycle. Concurrent stops on clones serialize behind one settlement and each returns only after it.

One reentrant edge degrades to signal-only: user code reached through a user-installed global logger runs ON the server threads, and if such code calls stop(), joining from there would deadlock — that call silently skips the joins (silently, because logging from inside the logger’s own frame could deadlock on the logger’s lock), leaves them available to any other caller, and returns with only the signal sent.

More generally: the joined threads log through the global logger while they wind down, so blocking in stop() while holding ANY resource that logger may acquire is a lock inversion. That has two shapes, neither detectable from this crate:

do not call stop() from inside a logger implementation (on any thread) — a log() blocking here while holding the logger’s own internal lock deadlocks against the winding-down threads’ log calls;
do not call stop() while holding a lock your logger sink also takes (e.g. an Arc<Mutex<_>> shared between application code and a custom logger) — the server threads block on that lock in their shutdown logging while stop() blocks joining them.

The same discipline applies to joining any thread that logs; the server-thread reentrant shape above is the only one the crate can detect cheaply, and it is guarded.

An FFmpeg job still pushing to this server (via create_rtmp_input) loses its feed and fails its next write; the server classifies that failure — best effort, by the server’s terminal cause — as a deliberate stop rather than as an opaque send error. A feed that already died for its own protocol error before the stop keeps the warning it got at removal time. Stop such jobs first if their clean completion matters.

§Example

let server = EmbedRtmpServer::new("localhost:1935");
// ... start and handle streaming
server.stop();
assert!(server.is_stopped());

Source §

impl EmbedRtmpServer<Initialization>

Source

pub fn stream_builder() -> StreamBuilder

Creates a new StreamBuilder for simplified RTMP streaming.

This is the recommended entry point for simple streaming scenarios where you want to stream a file to an embedded RTMP server.

§Example

use ez_ffmpeg::rtmp::embed_rtmp_server::EmbedRtmpServer;

let handle = EmbedRtmpServer::stream_builder()
    .address("localhost:1935")
    .app_name("live")
    .stream_key("stream1")
    .input_file("video.mp4")
    .start()?;

handle.wait()?;

For more complex scenarios requiring full control over the server and FFmpeg context, use the traditional API:

let server = EmbedRtmpServer::new("localhost:1935").start()?;
let output = server.create_rtmp_input("app", "stream")?;
// ... configure Input and FfmpegContext manually