Implement FCPublish/releaseStream, AMF3 shared objects, HDR colorInfo, exvideo/exaudio write, and E-RTMP v2 reconnect - #215
Conversation
The status tables and state-machine notes had drifted from the code and understated what's wired into the live session path: pause/seek/receiveAudio/ receiveVideo/closeStream, deleteStream teardown, onMetaData relay, User Control messages, exvideo/exaudio parsing, fourCcList echo, E-RTMP v2 caps negotiation (CAPS_NEGOTIATED is actually entered), and multitrack/ModEx are all implemented in session/conn.rs and server/mod.rs, not just library code. Also fixed the repo structure listing (amf/ and chunk/ are directories, not files; media/ and net.rs were missing).
…orInfo, exvideo/exaudio write helpers, and E-RTMP v2 reconnect Wires five previously-missing pieces into the live session path: - releaseStream now force-releases a stale publish-route claim so a reconnecting encoder can immediately republish (FCPublish stays a no-op). - AMF3 Shared Object messages (RTMP type 0x10) are parsed into a structured envelope (name/version/flags + event list) and delivered via a new on_shared_object_cb; Conn::send_shared_object writes them back out. Event payload contents are left opaque -- interpreting them is a host policy decision, consistent with this crate's design. - HDR/colorInfo parsing is wired into video frame population: an enhanced Metadata packet type now populates Frame.hdr / Frame.has_hdr. - exvideo_write / exaudio_write mirror the existing parse functions in reverse, for embedders that need to build enhanced tag headers. - Reconnect is now sent/received in-session: Conn::send_reconnect_request and Server::request_reconnect send NetConnection.Connect.ReconnectRequest; the client detects it mid-poll and fires on_reconnect_request_cb with the optional tcUrl. Establishing the new connection is left to the host. Frame gains two ABI-additive fields (has_hdr, hdr) appended at the end of the repr(C) struct per docs/abi-policy.md. README implementation status updated to match. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015ZRH22k5s8VSoSJgUXa2wd
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (12)
📝 WalkthroughWalkthroughThe PR adds shared-object parsing and writing, E-RTMP reconnect signaling, enhanced media header writers, HDR metadata extraction, and authorized ChangesProtocol message and shared-object support
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Server
participant Conn
participant Client
participant ReconnectCallback
Server->>Conn: request_reconnect
Conn->>Client: E-RTMP reconnect request
Client->>ReconnectCallback: invoke with optional tcUrl and description
Possibly related PRs
Suggested labels: Suggested reviewers: ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
cargo-semver-checks (CI) correctly flagged the previous commit: Frame is explicitly documented in docs/abi-policy.md as ABI-stable across minor/patch releases, so appending has_hdr/hdr fields to it was a real breaking change, not the "ABI-additive" change the commit claimed. Moves HDR delivery to a new Rust-only Conn::detected_hdr_info field instead (same pattern as the existing detected_video_codec/detected_audio_codec), populated from a new media::parse_video_metadata_hdr() helper. Frame is now byte-for-byte identical to the pre-change struct. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015ZRH22k5s8VSoSJgUXa2wd
…ntation-status-8e2sfa # Conflicts: # README.md
|
@codex review |
|
@coderabbitai review |
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1e6981c8ca
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if let Some(ref routes) = self.publish_routes { | ||
| routes.force_release(&self.app, &name_str); | ||
| } |
There was a problem hiding this comment.
When a second authenticated connection sends releaseStream for a route that is still actively published, this unconditionally removes the first connection's ownership without verifying that it is stale or belongs to the caller. The second connection can then claim the route while the original publisher continues relaying under the same key, breaking the server's single-publisher invariant and interleaving both feeds and cached initialization data. Restrict release to the owning connection, or confirm that the recorded owner is no longer live before force-releasing it.
Useful? React with 👍 / 👎.
| /// Fired when the server sends `NetConnection.Connect.ReconnectRequest` | ||
| /// (E-RTMP v2 reconnect mechanism). `tc_url` is `Some` when the server | ||
| /// wants the client to reconnect to a different URL, `None` to reuse the | ||
| /// current one. The library only delivers the event -- establishing the | ||
| /// new connection and disconnecting from this one is left to the host | ||
| /// application, which can keep streaming through the next media boundary | ||
| /// before doing so. | ||
| pub on_reconnect_request_cb: Option<fn(tc_url: Option<&str>, description: Option<&str>)>, |
There was a problem hiding this comment.
For servers that send reconnect requests only after negotiating E-RTMP v2 reconnect support, this callback can never fire: Client::do_amf_connect still passes None to build_connect, so the client advertises neither capsEx's reconnect bit nor the reconnect value. The loopback test bypasses this by calling Server::request_reconnect unconditionally, but a negotiation-compliant server will see a legacy client and never issue the request. The client needs to offer reconnect capabilities when this handler is enabled (or through an explicit capability configuration).
Useful? React with 👍 / 👎.
| if hdr.is_ex_header == 0 { | ||
| if buf.is_empty() { | ||
| return 0; | ||
| } | ||
| buf[0] = (hdr.frame_type & 0x0F) << 4; | ||
| return 1; |
There was a problem hiding this comment.
When is_ex_header == 0, the generated FLV video header contains only the frame type and always leaves the low codec-ID nibble as zero. Codec ID 0 is not a valid legacy video codec, so every legacy header produced by this branch is unusable even though the function reports success; the round-trip test misses this because VideoHeader does not retain the parsed legacy codec. Accept or retain a legacy codec ID and encode it here, or reject legacy mode instead of emitting an invalid header.
Useful? React with 👍 / 👎.
| for event in &msg.events { | ||
| if event.data.len() > MAX_SO_EVENT_DATA_BYTES { |
There was a problem hiding this comment.
For a SharedObjectMessage containing more than 256 events, write succeeds and Conn::send_shared_object transmits the result, but this module's own parse rejects that same message at MAX_SO_EVENTS. This makes the public writer capable of producing messages that local receivers silently drop, for example when a host batches 257 attribute changes. Validate msg.events.len() against MAX_SO_EVENTS before serializing.
Useful? React with 👍 / 👎.
| let mut color_info = crate::types::HdrInfo::default(); | ||
| metadata_colorinfo_parse(&payload[hdr.header_size..], &mut color_info).ok()?; | ||
| Some(color_info) |
There was a problem hiding this comment.
For actual Enhanced RTMP PacketType.Metadata packets, the bytes after the ExVideoTagHeader are AMF metadata containing the colorInfo object, not a six-byte packed primaries/transfer/matrix tuple. Passing the payload prefix directly to metadata_colorinfo_parse therefore interprets AMF type markers and key bytes as color values and usually returns plausible-looking but incorrect detected_hdr_info. Parse the AMF metadata structure and extract the nested color fields before populating HdrInfo.
Useful? React with 👍 / 👎.
| let mut name_bytes = vec![0u8; name_len]; | ||
| buf.read(&mut name_bytes).map_err(|_| ErrorCode::Amf)?; | ||
| let name = String::from_utf8_lossy(&name_bytes).into_owned(); |
There was a problem hiding this comment.
When a peer sends a shared-object name containing invalid UTF-8, from_utf8_lossy accepts it and replaces invalid sequences with U+FFFD. Distinct wire names can consequently collapse to the same SharedObjectMessage::name, which is unsafe when the host uses that name for the attribute-sync or persistence policy this API delegates to it. Since the envelope documents the name as UTF-8 and does not expose the original bytes, reject invalid encoding instead of normalizing it lossily.
Useful? React with 👍 / 👎.
| let Ok(so) = shared_object::parse(payload) else { | ||
| return Ok(()); | ||
| }; | ||
| if let Some(cb) = self.on_shared_object_cb { | ||
| cb(self.conn_id, &so); |
There was a problem hiding this comment.
After the RTMP handshake but before any connect command, the connection is already in ConnState::Connected and read_messages dispatches type 0x10 messages here without an application or authorization check. A peer can therefore invoke the host's shared-object synchronization or persistence callback while self.app is still empty and before any connect-time policy has run; the callback receives only a connection ID and message, so it cannot reliably recover the missing application namespace. Ignore or reject shared-object messages until the connection reaches AppConnected.
Useful? React with 👍 / 👎.
| /// Bit 0 = persistent, per spec. | ||
| pub flags: u32, | ||
| pub events: Vec<SharedObjectEvent>, | ||
| } | ||
|
|
||
| impl SharedObjectMessage { | ||
| pub fn is_persistent(&self) -> bool { | ||
| self.flags & 0x01 != 0 |
There was a problem hiding this comment.
RTMP shared-object persistence is represented by flag value 0x02 (the second bit), but is_persistent tests 0x01 and the new tests construct that nonstandard value. Consequently a standard incoming persistent shared object is reported as non-persistent, while messages created with this API's documented flag are not recognized as persistent by conforming peers. Test and document 0x02 instead.
Useful? React with 👍 / 👎.
- media/init_cache.rs: parse colorInfo from its actual AMF0 metadata object (colorInfo.colorConfig or a bare colorConfig, per the Enhanced RTMP v1 spec) instead of a raw 6-byte tuple, which was misreading AMF type/key bytes as color values. - session/conn.rs: releaseStream no longer force-evicts another connection's live publish-route claim by default -- gated behind a new on_release_stream_cb that must explicitly authorize it, since the previous unconditional eviction let any authenticated peer hijack another stream by name. - session/conn.rs: AMF3 shared-object messages are now rejected before the connect command completes, instead of being dispatched to on_shared_object_cb while self.app is still empty and unauthorized. - message/shared_object.rs: reject invalid UTF-8 shared-object names instead of lossily substituting them (distinct wire names must not collapse onto the same host-visible name); write() now enforces the same MAX_SO_EVENTS cap parse() does, so it can't produce a message parse() would reject; softened the persistence-flag doc comment since the exact bit has not been independently verified. - ertmp/exvideo.rs: exvideo_write's legacy (non-ex-header) branch always emitted codec ID 0, which is not a valid legacy video codec -- VideoHeader has no field to carry the real one, so legacy mode now returns 0 (unsupported) instead of emitting an invalid tag byte. - client/mod.rs: Client::connect now advertises the E-RTMP v2 reconnect capability (capsEx bit + a reconnect value) on the wire whenever on_reconnect_request_cb is set, so a spec-compliant server has a reason to actually send a ReconnectRequest to this client. README updated to match. All new/changed behavior covered by unit and loopback integration tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015ZRH22k5s8VSoSJgUXa2wd
The new client_with/without_reconnect_callback_advertises_reconnect_capability tests were flaking in CI: the driver loop's exit condition was based on an inferred server-side connection state, and its wall-clock deadline could elapse before the client's own connect() finished on a contended runner -- observed both as a plain timeout and, in one run, as the server never even having accepted the connection by the time the loop gave up. Switch to the same pattern already used successfully elsewhere in this file: keep polling until the client thread itself reports completion via the setup channel, with a deadline that comfortably outlives the client's own (now explicit) connect timeout instead of racing a shorter one. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015ZRH22k5s8VSoSJgUXa2wd
Found the actual root cause of the remaining CI flake: `client` was constructed and dropped inside the innermost closure, so it closed its TCP connection immediately after connect() returned Ok -- before (or racing) the setup_tx notification the driver loop was waiting on. If the driver loop's next server.poll() ran before it saw that notification, it could observe and process the resulting disconnect, evicting the connection from server.connections before the test ever got to assert on it (reproduced in CI as `left: 0, right: 1` in ~1.8s, i.e. not a timeout). Move `client` out to the thread closure's own scope so it stays connected through the success signal and the driver loop's final break -- no poll() call happens after that point, so the connection can no longer be raced away. Verified stable over 10 runs locally, both with default and --test-threads=1 parallelism (the mode the failing CI job used). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015ZRH22k5s8VSoSJgUXa2wd
|
ⓘ Qodo reviews are paused because your trial has ended. Ask your workspace admin to add credits to resume reviews. Manage billing |
2 participants
Summary
Follow-up to #214 (README implementation-status audit). This implements the five items that audit flagged as genuinely missing:
FCPublish/releaseStream—releaseStreamnow force-releases a stale publish-route claim (PublishRouteRegistry::force_release) so a reconnecting encoder can immediately republish without waiting out the old claim's timeout.FCPublishstays a no-op (matches its notification-only real-world semantics).src/message/shared_object.rsparses/writes the RTMP Shared Object envelope (name/version/flags + event list, per Adobe RTMP 1.0 §7.1) with bounds-checked, never-trust-the-wire parsing.session/conn.rswires this intohandle_amf3_shared_object, delivering parsed messages via a newon_shared_object_cb;Conn::send_shared_objectwrites them back out. Event payload contents (e.g.Change/SendMessagevalues) are left as opaque bytes — interpreting them is host policy, consistent with this crate's "deliver the event, not the policy" design.colorInfo—metadata::metadata_colorinfo_parse(previously dead code) is now called via a newmedia::parse_video_metadata_hdr()helper whenever an enhanced video frame's packet type isMetadata, exposed via a newConn::detected_hdr_infofield (same pattern as the existingdetected_video_codec/detected_audio_codec). Kept offFramedeliberately —docs/abi-policy.mddocumentsFrame's#[repr(C)]layout as ABI-stable across minor/patch releases, and CI'scargo-semver-checkscorrectly caught an earlier draft of this PR that appended fields to it.exvideo_write/exaudio_write— mirror the existingexvideo_parse/exaudio_parsein reverse, for embedders building enhanced tag headers.Conn::send_reconnect_request/Server::request_reconnectsendNetConnection.Connect.ReconnectRequest(E-RTMP v2 reconnect mechanism); the client now processes AMF commands received outside the connect/publish/play handshake and fires a newon_reconnect_request_cbwith the optionaltcUrl. Establishing the new connection is left to the host application, matching this library's "protocol, not policy" scope.README's Implementation status tables are updated to match.
Test plan
cargo build/cargo build --no-default-featurescargo test(396 lib tests + all integration tests, including a new end-to-end loopback test asserting a server-sent reconnect request reaches the client's callback with the righttcUrl)cargo clippy --all-features --all-targetscleancargo fmt --checkclean on all touched files (pre-existing formatting drift in untouched files/lines left as-is)cargo-semver-checks(CI) —Frame's public field set is unchanged from the previous releaseexvideo/exaudiowrite round-trips,releaseStreamforce-release, HDR/colorInfo detection, reconnect-request build/parse + client callback wiring, shared-object envelope parse/write + session dispatchSummary by CodeRabbit
New Features
releaseStreamhandling to release stale publishing routes safely.Documentation