starred/open-bamboo-networking
1
0 Fork 0
mirror of https://github.com/ClusterM/open-bamboo-networking.git synced 2026-10-19 01:29:46 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

Implement deadline-aware SSL read functions for RTSP client (#74)

Browse Source 
* Implement deadline-aware SSL read functions for RTSP client

- Added `ssl_read_full_timeout` and `ssl_read_line_timeout` to handle timeouts during SSL reads, preventing indefinite blocking on read operations.
- Updated `read_response` in `rtsp_client.cpp` to utilize the new timeout functions, ensuring that responses are read within specified deadlines.
- Enhanced error handling for RTSP sessions to log errors when a stream dies mid-session.
- Introduced `SpsGeometry` struct and `parse_sps_geometry` function to decode display geometry from SPS NAL units, improving video stream handling.
- Updated `BambuSource` and `Passthrough` implementations to utilize the new geometry parsing for better video parameter management.

Signed-off-by: Alexey Cluster <cluster@cluster.wtf>
This commit is contained in:
Alexey Cluster authored and GitHub committed 2026-08-11 21:34:54 +02:00
1 parent 8c6b771376
commit 5ef86b200b
10 files changed
+1003 -82
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

include/obn/tls_dial.hpp
+16
View File
@@ -39,4 +39,20 @@ int ssl_read_full(SSL* ssl, void* buf, std::size_t len);
int ssl_read_line(SSL* ssl, std::string* out, std::size_t max_len = 8192);
// Deadline-aware variants of the two readers above. `timeout_ms` bounds
// the wait for *each* chunk of incoming data, so a peer that goes quiet
// mid-message fails with -1 and a "read timed out" last_error instead of
// parking the calling thread in SSL_read forever. A long-lived stream
// reader (the RTSP data plane) needs this: without it a silently dropped
// session is indistinguishable from an idle one.
//
// The bound holds on its own: both apply the budget to the descriptor's
// receive timeout for the duration of the call, so a half-delivered TLS
// record cannot block past the deadline and callers need not configure
// the socket themselves.
int ssl_read_full_timeout(SSL* ssl, void* buf, std::size_t len, int timeout_ms);
int ssl_read_line_timeout(SSL* ssl, std::string* out, int timeout_ms,
std::size_t max_len = 8192);
} // namespace obn::tls
src/tls_dial.cpp
+216
View File
@@ -21,6 +21,7 @@
# include <ws2tcpip.h>
# include <windows.h>
#else
# include <csignal>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/tcp.h>
@@ -29,7 +30,9 @@
# include <unistd.h>
#endif
#include <cerrno>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <mutex>
@@ -51,10 +54,31 @@ void set_error(const char* msg)
}
}
// A write to a socket whose peer is gone raises SIGPIPE, and the default
// disposition kills the whole process -- here, the slicer that loaded us.
// OpenSSL has no per-write MSG_NOSIGNAL equivalent, so neutralise the
// signal instead, and only when the host has not installed a handler of
// its own (never override the application's choice).
void ignore_sigpipe_if_default()
{
#if !defined(_WIN32)
struct sigaction old{};
if (::sigaction(SIGPIPE, nullptr, &old) != 0) return;
if (old.sa_handler != SIG_DFL) return;
struct sigaction act{};
act.sa_handler = SIG_IGN;
// Unqualified on purpose: Apple's SDK defines sigemptyset as a macro,
// which a ::-qualified call cannot name.
sigemptyset(&act.sa_mask);
::sigaction(SIGPIPE, &act, nullptr);
#endif
}
void init_once()
{
std::call_once(g_init_flag, []() {
obn::os::winsock_init_once();
ignore_sigpipe_if_default();
SSL_library_init();
OpenSSL_add_all_algorithms();
@@ -368,6 +392,198 @@ int ssl_read_full(SSL* ssl, void* buf, std::size_t len)
return 0;
}
namespace {
using SteadyPoint = std::chrono::steady_clock::time_point;
// Block until the socket can move data in the direction OpenSSL asked for
// (`want` is poll_event::in or ::out) or `deadline` passes. Returns 1 when
// the next SSL call can make progress, 0 on timeout, -1 on a dead socket.
// SSL_pending() short-circuits the poll: bytes already sitting in the
// record buffer are readable even when the socket is not.
int wait_ssl_io(SSL* ssl, short want, SteadyPoint deadline)
{
if (want == obn::net::poll_event::in && SSL_pending(ssl) > 0) return 1;
const obn::os::socket_t fd =
static_cast<obn::os::socket_t>(SSL_get_fd(ssl));
if (!obn::os::socket_valid(fd)) return -1;
for (;;) {
const auto now = std::chrono::steady_clock::now();
if (now >= deadline) return 0;
auto left_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
deadline - now).count();
// Cap a single poll so a caller that shuts the socket down from
// another thread is noticed promptly even on a long deadline.
if (left_ms > 1000) left_ms = 1000;
short revents = 0;
const int rc = obn::os::poll_one(fd, want,
static_cast<int>(left_ms), &revents);
if (rc < 0) return -1;
if (rc > 0) return (revents & want) ? 1 : -1;
}
}
void set_timeout_error(int timeout_ms, short want)
{
char msg[96];
std::snprintf(msg, sizeof(msg), "%s timed out after %d ms",
(want == obn::net::poll_event::out) ? "TLS write" : "read",
timeout_ms);
set_error(msg);
}
// SSL_read on a blocking BIO parks inside recv() until the peer delivers
// the rest of a TLS record it started -- polling for readability first
// cannot prevent that, it only proves the record's opening bytes arrived.
// Bound the syscall itself for the duration of a timed read so the
// deadline holds however the caller left the descriptor; dial_tls clears
// its handshake timeout, so relying on the caller would make the timeout
// helpers silently unbounded. read_some_timeout reads the resulting EAGAIN
// as "not ready yet" and goes back to its own deadline check.
//
// A receive timeout already at least as tight as our budget is left alone,
// which keeps the common path (the RTSP client sets one) syscall-free.
class ScopedRecvTimeout {
public:
ScopedRecvTimeout(obn::os::socket_t fd, int timeout_ms) : fd_(fd)
{
if (!obn::os::socket_valid(fd_) || timeout_ms <= 0) return;
#if defined(_WIN32)
int len = static_cast<int>(sizeof(prev_));
if (::getsockopt(static_cast<SOCKET>(fd_), SOL_SOCKET, SO_RCVTIMEO,
reinterpret_cast<char*>(&prev_), &len) != 0)
return;
const DWORD want = static_cast<DWORD>(timeout_ms);
if (prev_ != 0 && prev_ <= want) return;
if (::setsockopt(static_cast<SOCKET>(fd_), SOL_SOCKET, SO_RCVTIMEO,
reinterpret_cast<const char*>(&want),
sizeof(want)) == 0)
restore_ = true;
#else
socklen_t len = static_cast<socklen_t>(sizeof(prev_));
if (::getsockopt(fd_, SOL_SOCKET, SO_RCVTIMEO, &prev_, &len) != 0)
return;
const long prev_ms = static_cast<long>(prev_.tv_sec) * 1000 +
prev_.tv_usec / 1000;
if (prev_ms > 0 && prev_ms <= timeout_ms) return;
timeval want{};
want.tv_sec = timeout_ms / 1000;
want.tv_usec = (timeout_ms % 1000) * 1000;
if (::setsockopt(fd_, SOL_SOCKET, SO_RCVTIMEO, &want,
sizeof(want)) == 0)
restore_ = true;
#endif
}
~ScopedRecvTimeout()
{
if (!restore_) return;
#if defined(_WIN32)
::setsockopt(static_cast<SOCKET>(fd_), SOL_SOCKET, SO_RCVTIMEO,
reinterpret_cast<const char*>(&prev_), sizeof(prev_));
#else
::setsockopt(fd_, SOL_SOCKET, SO_RCVTIMEO, &prev_, sizeof(prev_));
#endif
}
ScopedRecvTimeout(const ScopedRecvTimeout&) = delete;
ScopedRecvTimeout& operator=(const ScopedRecvTimeout&) = delete;
private:
obn::os::socket_t fd_;
bool restore_ = false;
#if defined(_WIN32)
DWORD prev_ = 0;
#else
timeval prev_{};
#endif
};
// Read exactly one chunk into `buf`, waiting no longer than `timeout_ms`
// for it to start arriving. Returns the byte count (>0), 0 for a clean
// EOS, or -1 on error / timeout.
int read_some_timeout(SSL* ssl, void* buf, std::size_t len, int timeout_ms)
{
const auto deadline = std::chrono::steady_clock::now() +
std::chrono::milliseconds(timeout_ms);
short want = obn::net::poll_event::in;
for (;;) {
const int ready = wait_ssl_io(ssl, want, deadline);
if (ready == 0) {
set_timeout_error(timeout_ms, want);
return -1;
}
if (ready < 0) return -1;
const int n = SSL_read(ssl, buf, static_cast<int>(len));
if (n > 0) return n;
const int err = SSL_get_error(ssl, n);
if (err == SSL_ERROR_ZERO_RETURN) return 0;
// A read that has to send first (renegotiation, TLS 1.3 KeyUpdate)
// needs the socket writable, not readable -- polling the wrong
// direction would burn the whole budget and report a phantom read
// timeout. Blocking BIOs rarely surface this, but SO_SNDTIMEO
// expiring mid-flight does.
if (err == SSL_ERROR_WANT_WRITE) {
want = obn::net::poll_event::out;
continue;
}
if (err == SSL_ERROR_WANT_READ) {
want = obn::net::poll_event::in;
continue; // short record: wait for the rest
}
if (err == SSL_ERROR_SYSCALL &&
(obn::os::socket_in_progress(obn::os::last_socket_error()) ||
obn::os::last_socket_error() == EINTR))
continue; // SO_RCVTIMEO / signal, not a real failure
return -1;
}
}
} // namespace
int ssl_read_full_timeout(SSL* ssl, void* buf, std::size_t len, int timeout_ms)
{
if (!ssl) return -1;
if (timeout_ms <= 0) return ssl_read_full(ssl, buf, len);
ScopedRecvTimeout guard(static_cast<obn::os::socket_t>(SSL_get_fd(ssl)),
timeout_ms);
auto* p = static_cast<std::uint8_t*>(buf);
std::size_t got = 0;
while (got < len) {
const int n = read_some_timeout(ssl, p + got, len - got, timeout_ms);
if (n == 0) return 1;
if (n < 0) return -1;
got += static_cast<std::size_t>(n);
}
return 0;
}
int ssl_read_line_timeout(SSL* ssl, std::string* out, int timeout_ms,
std::size_t max_len)
{
if (!ssl) return -1;
if (timeout_ms <= 0) return ssl_read_line(ssl, out, max_len);
ScopedRecvTimeout guard(static_cast<obn::os::socket_t>(SSL_get_fd(ssl)),
timeout_ms);
out->clear();
out->reserve(128);
char prev = '\0';
while (out->size() < max_len) {
char c = '\0';
const int n = read_some_timeout(ssl, &c, 1, timeout_ms);
if (n == 0) return 1;
if (n < 0) return -1;
if (prev == '\r' && c == '\n') {
if (!out->empty()) out->pop_back();
return 0;
}
out->push_back(c);
prev = c;
}
set_error("ssl_read_line_timeout: line exceeded max_len");
return -1;
}
int ssl_read_line(SSL* ssl, std::string* out, std::size_t max_len)
{
out->clear();
stubs/BambuSource.cpp
+13 -8
View File
@@ -666,17 +666,22 @@ int ssl_read_all(Tunnel* t, void* buf, size_t len)
// pipeline already speaks (h264parse copes with any framing
// h264parse can detect, and Annex-B is the simplest one).
t->sub_type = AVC1;
// Width/height/frame_rate are advisory until h264parse pulls them
// out of SPS; surface the firmware's well-known 1280x720@30 default
// so Studio's UI shows reasonable numbers from the start.
t->width = 1280;
t->height = 720;
t->frame_rate = 30;
// Prefer what the SDP's SPS actually says: the coded size and frame
// rate differ per model (a P2S serves 1168x720 @ ~24.7 fps), and a
// wrong frame rate makes the slicer's sink judge live frames late.
// The 1280x720@30 fallback only applies when the SPS did not parse.
{
const auto p = t->rtsp_pass->params();
t->width = (p.width > 0) ? p.width : 1280;
t->height = (p.height > 0) ? p.height : 720;
t->frame_rate = (p.fps > 0) ? p.fps : 30;
}
t->t0 = std::chrono::steady_clock::now();
t->started = true;
log_fmt(t->logger, t->log_ctx,
"open_rtsp: passthrough ready (avc1 %dx%d, gstbambusrc decodes)",
t->width, t->height);
"open_rtsp: passthrough ready (avc1 %dx%d @ %d fps, "
"gstbambusrc decodes)",
t->width, t->height, t->frame_rate);
return Bambu_success;
}
stubs/h264_avcc.cpp
+191
View File
@@ -2,6 +2,7 @@
#include <cstdint>
#include <limits>
#include <vector>
namespace obn {
namespace h264 {
@@ -77,6 +78,89 @@ bool append_avcc_nal(std::vector<uint8_t>& out, const uint8_t* nal, size_t nal_s
return true;
}
// Bit reader over an RBSP: the emulation-prevention 0x03 bytes are
// stripped up front so the SPS syntax elements can be read straight out
// of the bit stream. Every accessor is bounds-checked and latches `ok_`
// on overrun, so a truncated SPS fails the parse instead of reading past
// the buffer.
class RbspReader {
public:
RbspReader(const uint8_t* data, size_t n)
{
buf_.reserve(n);
for (size_t i = 0; i < n; ++i) {
if (i + 2 < n && data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 3) {
buf_.push_back(0);
buf_.push_back(0);
i += 2;
continue;
}
buf_.push_back(data[i]);
}
}
uint32_t u(int bits)
{
uint32_t v = 0;
for (int i = 0; i < bits; ++i) {
const size_t byte = bit_ >> 3;
if (byte >= buf_.size()) { ok_ = false; return 0; }
const int shift = 7 - static_cast<int>(bit_ & 7);
v = (v << 1) | ((buf_[byte] >> shift) & 1u);
++bit_;
}
return v;
}
uint32_t ue()
{
int zeros = 0;
while (ok_ && u(1) == 0) {
if (++zeros > 31) { ok_ = false; return 0; }
}
if (!ok_ || zeros == 0) return 0;
return (1u << zeros) - 1 + u(zeros);
}
int32_t se()
{
const uint32_t k = ue();
if (k == 0) return 0;
const int32_t mag = static_cast<int32_t>((k + 1) / 2);
return (k & 1) ? mag : -mag;
}
bool ok() const { return ok_; }
private:
std::vector<uint8_t> buf_;
size_t bit_ = 0;
bool ok_ = true;
};
// True for the profiles whose SPS carries the chroma_format_idc block
// (Annex A high profiles). Bambu encodes High, so this branch matters.
bool profile_has_chroma_block(int profile_idc)
{
switch (profile_idc) {
case 100: case 110: case 122: case 244:
case 44: case 83: case 86: case 118:
case 128: case 138: case 139: case 134: case 135:
return true;
default:
return false;
}
}
void skip_scaling_list(RbspReader& r, int size)
{
int32_t last = 8, next = 8;
for (int i = 0; i < size && r.ok(); ++i) {
if (next != 0) next = (last + r.se() + 256) % 256;
last = (next == 0) ? last : next;
}
}
} // namespace
bool extract_parameter_sets(const uint8_t* data, size_t size, ParameterSets* out)
@@ -123,5 +207,112 @@ bool contains_idr(const uint8_t* data, size_t size)
return found;
}
bool parse_sps_geometry(const uint8_t* sps, size_t size, SpsGeometry* out)
{
if (!sps || !out || size < 4) return false;
RbspReader r(sps + 1, size - 1); // skip the NAL header
const int profile_idc = static_cast<int>(r.u(8));
r.u(8); // constraint flags + reserved
r.u(8); // level_idc
r.ue(); // seq_parameter_set_id
uint32_t chroma_format_idc = 1; // 4:2:0 unless stated
bool separate_colour_plane_flag = false;
if (profile_has_chroma_block(profile_idc)) {
chroma_format_idc = r.ue();
if (chroma_format_idc == 3) separate_colour_plane_flag = r.u(1) != 0;
r.ue(); // bit_depth_luma_minus8
r.ue(); // bit_depth_chroma_minus8
r.u(1); // qpprime_y_zero_transform_bypass
if (r.u(1)) { // seq_scaling_matrix_present
const int lists = (chroma_format_idc != 3) ? 8 : 12;
for (int i = 0; i < lists && r.ok(); ++i) {
if (r.u(1)) skip_scaling_list(r, i < 6 ? 16 : 64);
}
}
}
r.ue(); // log2_max_frame_num_minus4
const uint32_t poc_type = r.ue();
if (poc_type == 0) {
r.ue(); // log2_max_poc_lsb_minus4
} else if (poc_type == 1) {
r.u(1); // delta_pic_order_always_zero
r.se(); // offset_for_non_ref_pic
r.se(); // offset_for_top_to_bottom_field
const uint32_t n = r.ue();
for (uint32_t i = 0; i < n && r.ok(); ++i) r.se();
}
r.ue(); // max_num_ref_frames
r.u(1); // gaps_in_frame_num_allowed
const uint32_t width_mbs = r.ue() + 1;
const uint32_t height_maps = r.ue() + 1;
const bool frame_mbs_only = r.u(1) != 0;
if (!frame_mbs_only) r.u(1); // mb_adaptive_frame_field
r.u(1); // direct_8x8_inference
uint32_t crop_l = 0, crop_r = 0, crop_t = 0, crop_b = 0;
if (r.u(1)) { // frame_cropping_flag
crop_l = r.ue();
crop_r = r.ue();
crop_t = r.ue();
crop_b = r.ue();
}
int fps = 0;
if (r.u(1)) { // vui_parameters_present
if (r.u(1)) { // aspect_ratio_info_present
if (r.u(8) == 255) { r.u(16); r.u(16); } // sar_width / sar_height
}
if (r.u(1)) r.u(1); // overscan
if (r.u(1)) { // video_signal_type_present
r.u(3); // video_format
r.u(1); // video_full_range
if (r.u(1)) { r.u(8); r.u(8); r.u(8); } // colour description
}
if (r.u(1)) { r.ue(); r.ue(); } // chroma_loc_info
if (r.u(1)) { // timing_info_present
const uint32_t num_units_in_tick = r.u(32);
const uint32_t time_scale = r.u(32);
r.u(1); // fixed_frame_rate_flag
if (r.ok() && num_units_in_tick > 0 && time_scale > 0) {
// H.264 ties one tick to a field, so a frame takes two.
const double v = static_cast<double>(time_scale) /
(2.0 * static_cast<double>(num_units_in_tick));
if (v > 0.0 && v < 1000.0) fps = static_cast<int>(v + 0.5);
}
}
}
if (!r.ok()) return false;
// Cropping offsets count chroma sample units (spec 7.4.2.1.1).
const uint32_t chroma_array_type =
separate_colour_plane_flag ? 0 : chroma_format_idc;
uint32_t sub_w = 1, sub_h = 1;
switch (chroma_array_type) {
case 1: sub_w = 2; sub_h = 2; break; // 4:2:0
case 2: sub_w = 2; sub_h = 1; break; // 4:2:2
default: break; // 4:4:4 / monochrome
}
const uint32_t unit_x = (chroma_array_type == 0) ? 1 : sub_w;
const uint32_t unit_y = ((chroma_array_type == 0) ? 1 : sub_h) *
(frame_mbs_only ? 1u : 2u);
const int64_t w = static_cast<int64_t>(width_mbs) * 16 -
static_cast<int64_t>(crop_l + crop_r) * unit_x;
const int64_t h = static_cast<int64_t>(height_maps) * 16 *
(frame_mbs_only ? 1 : 2) -
static_cast<int64_t>(crop_t + crop_b) * unit_y;
if (w <= 0 || h <= 0 || w > 16384 || h > 16384) return false;
out->width = static_cast<int>(w);
out->height = static_cast<int>(h);
out->fps = fps;
return true;
}
} // namespace h264
} // namespace obn
stubs/h264_avcc.hpp
+14
View File
@@ -37,5 +37,19 @@ bool annexb_to_avcc(const uint8_t* data, size_t size, AvccFrame* out);
// True if the Annex-B buffer contains an IDR slice (NAL type 5).
bool contains_idr(const uint8_t* data, size_t size);
// Display geometry carried by an SPS. `fps` is 0 when the SPS omits VUI
// timing information, which is optional.
struct SpsGeometry {
int width = 0;
int height = 0;
int fps = 0;
};
// Decode display geometry (cropping applied) and the frame rate from a
// raw SPS NAL -- `sps` includes the 1-byte NAL header but no Annex-B
// start code. Returns false when the SPS does not parse, leaving *out
// untouched.
bool parse_sps_geometry(const uint8_t* sps, size_t size, SpsGeometry* out);
} // namespace h264
} // namespace obn
stubs/rtsp_client.cpp
+426 -71
View File
@@ -9,15 +9,27 @@
// from RFC 2326 section 10.12. The interleaved channels are picked at
// SETUP time -- we always ask for 0 = RTP, 1 = RTCP.
//
// The control plane keeps running after PLAY: GET_PARAMETER keepalive
// requests fire every 15 seconds on a worker thread to avoid Bambu's
// 30 s idle teardown. Their responses come back through the same TCP
// connection; we let them flow into the demux loop, which recognises
// them by the leading "RTSP/" string instead of the '$' interleave
// marker, drains the response, and continues.
// The control plane keeps running after PLAY, with two keepalives:
//
// * RTCP receiver reports on interleaved channel 1 every 2 s, emitted
// by the reader thread between messages. The printer's live555 RTP
// sink treats a receiver that never reports as gone and stops
// sending; the session then simply goes quiet with no RTSP error
// anywhere (issue #54).
// * RTSP GET_PARAMETER every 5 s from a dedicated thread, which
// refreshes the session timeout on the server. Its response comes
// back through the same TCP connection; the demux loop recognises
// it by the leading "RTSP/" string instead of the '$' interleave
// marker, drains it, and continues.
//
// Every read on this connection carries a deadline
// (ssl_read_full_timeout). A stream that dies mid-session therefore
// surfaces as a logged error instead of parking the reader thread in
// SSL_read forever.
#include "rtsp_client.hpp"
#include "h264_avcc.hpp"
#include "source_log.hpp"
#include "tls_socket.hpp"
@@ -49,6 +61,7 @@
#include <cstdarg>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <mutex>
@@ -76,6 +89,33 @@ using obn::source::set_last_error;
constexpr std::size_t kMaxResponseBody = 64u << 10;
constexpr std::size_t kMaxRtpPacket = 64u << 10;
// Keepalive cadence. The RTCP interval is what keeps the server's RTP
// sink convinced someone is listening; the RTSP ping only refreshes the
// session timeout, so it can be much lazier.
constexpr int kRtcpIntervalMs = 2000;
constexpr int kRtspPingMs = 5000;
// How long the data plane tolerates silence before declaring the
// session dead. Generous enough to ride out a stalled GOP, short enough
// that the UI does not sit on a frozen frame. OBN_RTSP_READ_TIMEOUT_MS
// overrides it for field diagnosis.
constexpr int kDataReadTimeoutMs = 5000;
int data_read_timeout_ms()
{
if (const char* e = std::getenv("OBN_RTSP_READ_TIMEOUT_MS")) {
const int v = std::atoi(e);
if (v > 0) return v;
}
return kDataReadTimeoutMs;
}
// RTCP payload types we care about (RFC 3550 section 12.1).
constexpr int kRtcpSr = 200;
constexpr int kRtcpSdes = 202;
constexpr int kRtcpBye = 203;
constexpr int kRtcpRr = 201;
// ---------- base64 (small, inline) ----------
constexpr char kB64Alphabet[] =
@@ -308,8 +348,9 @@ struct Response {
// `$` frames apart from `R` (RTSP/1.0 ...) responses, then hands the
// `R` here to be re-stitched into the status line.
//
// Returns 0 / 1 / -1 like ssl_read_full.
int read_response(SSL* ssl, Response* out, int prefetch_byte = -1)
// Returns 0 / 1 / -1 like ssl_read_full. `timeout_ms` bounds every read
// the response needs, so a half-sent reply cannot wedge the caller.
int read_response(SSL* ssl, Response* out, int timeout_ms, int prefetch_byte = -1)
{
out->headers.clear();
out->body.clear();
@@ -321,26 +362,12 @@ int read_response(SSL* ssl, Response* out, int prefetch_byte = -1)
// Read the rest of the status line, byte by byte, with the
// supplied byte as the seed. Mirrors ssl_read_line semantics.
line.push_back(static_cast<char>(prefetch_byte));
char prev = static_cast<char>(prefetch_byte);
while (line.size() < 8192) {
char c = '\0';
int n = SSL_read(ssl, &c, 1);
if (n <= 0) {
int err = SSL_get_error(ssl, n);
if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE)
continue;
if (err == SSL_ERROR_ZERO_RETURN) return 1;
return -1;
}
if (prev == '\r' && c == '\n') {
line.pop_back(); // strip the CR we just pushed
break;
}
line.push_back(c);
prev = c;
}
std::string rest;
int rc = obn::tls::ssl_read_line_timeout(ssl, &rest, timeout_ms);
if (rc != 0) return rc;
line += rest;
} else {
int rc = obn::tls::ssl_read_line(ssl, &line);
int rc = obn::tls::ssl_read_line_timeout(ssl, &line, timeout_ms);
if (rc != 0) return rc;
}
// "RTSP/1.0 200 OK"
@@ -359,7 +386,7 @@ int read_response(SSL* ssl, Response* out, int prefetch_byte = -1)
// Headers.
std::size_t content_length = 0;
for (;;) {
int hrc = obn::tls::ssl_read_line(ssl, &line);
int hrc = obn::tls::ssl_read_line_timeout(ssl, &line, timeout_ms);
if (hrc != 0) return hrc;
if (line.empty()) break;
auto colon = line.find(':');
@@ -376,7 +403,8 @@ int read_response(SSL* ssl, Response* out, int prefetch_byte = -1)
}
if (content_length > 0) {
out->body.resize(content_length);
int brc = obn::tls::ssl_read_full(ssl, out->body.data(), content_length);
int brc = obn::tls::ssl_read_full_timeout(ssl, out->body.data(),
content_length, timeout_ms);
if (brc != 0) return brc;
}
return 0;
@@ -393,6 +421,9 @@ bool parse_sdp(const std::string& sdp, H264Track* track, std::string* control)
track->pps.clear();
track->clock_rate = 90000;
track->rtp_pt = -1;
track->width = 0;
track->height = 0;
track->fps = 0;
control->clear();
int current_pt = -1;
@@ -489,6 +520,15 @@ bool parse_sdp(const std::string& sdp, H264Track* track, std::string* control)
set_last_error("rtsp: SDP had no usable H.264 video track");
return false;
}
if (!track->sps.empty()) {
obn::h264::SpsGeometry geom;
if (obn::h264::parse_sps_geometry(track->sps.data(), track->sps.size(),
&geom)) {
track->width = geom.width;
track->height = geom.height;
track->fps = geom.fps;
}
}
return true;
}
@@ -553,12 +593,55 @@ struct Client::Impl {
// Keepalive.
std::thread keepalive;
std::atomic<bool> stop_flag{false};
// Set once the socket has been shut down. Any write after that point
// can only draw an EPIPE, so the senders below bail out instead.
std::atomic<bool> socket_down{false};
std::condition_variable keepalive_cv;
std::mutex keepalive_mu;
// Read deadlines. The control plane inherits the connect timeout
// (a printer that accepts TCP but never answers DESCRIBE must not
// wedge start()); the data plane gets its own, longer budget.
int ctrl_timeout_ms = 5000;
int data_timeout_ms = kDataReadTimeoutMs;
// True once the server sent an RTCP BYE: the session ended by
// agreement, so the reader reports a clean EOS rather than an error.
bool peer_bye = false;
// Receiver-side RTP bookkeeping for the RTCP reports. Written by
// the data thread, read by the keepalive thread, hence the mutex.
// The counters follow RFC 3550 appendix A.3.
std::mutex stats_mu;
std::uint32_t our_ssrc = 0;
bool have_peer_ssrc = false;
std::uint32_t peer_ssrc = 0;
bool seq_initialised = false;
std::uint16_t max_seq = 0;
std::uint32_t seq_cycles = 0;
std::uint32_t base_seq = 0;
std::uint32_t packets_received = 0;
std::uint32_t expected_prior = 0;
std::uint32_t received_prior = 0;
double jitter = 0.0;
bool have_transit = false;
std::int64_t last_transit = 0;
bool have_sr = false;
std::uint32_t last_sr_middle32 = 0;
std::chrono::steady_clock::time_point last_sr_at{};
std::chrono::steady_clock::time_point stream_t0{};
// When the next receiver report is due. Reader thread only.
std::chrono::steady_clock::time_point next_rtcp_due{};
H264Track track;
Impl(Logger l, void* c) : logger(l ? l : obn::source::noop_logger), log_ctx(c) {}
Impl(Logger l, void* c) : logger(l ? l : obn::source::noop_logger), log_ctx(c)
{
std::random_device rd;
our_ssrc = (static_cast<std::uint32_t>(rd()) << 16) ^ rd();
stream_t0 = std::chrono::steady_clock::now();
}
~Impl() = default;
@@ -747,11 +830,12 @@ struct Client::Impl {
// frames into nalu_queue for the data path to drain later.
for (;;) {
std::uint8_t prefix;
int rc = obn::tls::ssl_read_full(ssl, &prefix, 1);
int rc = obn::tls::ssl_read_full_timeout(ssl, &prefix, 1,
ctrl_timeout_ms);
if (rc != 0) return rc;
if (prefix == '$') {
std::uint8_t hdr[3];
rc = obn::tls::ssl_read_full(ssl, hdr, 3);
rc = obn::tls::ssl_read_full_timeout(ssl, hdr, 3, ctrl_timeout_ms);
if (rc != 0) return rc;
int channel = hdr[0];
std::uint16_t plen = static_cast<std::uint16_t>(
@@ -762,20 +846,22 @@ struct Client::Impl {
}
std::vector<std::uint8_t> rtp(plen);
if (plen > 0) {
rc = obn::tls::ssl_read_full(ssl, rtp.data(), plen);
rc = obn::tls::ssl_read_full_timeout(ssl, rtp.data(), plen,
ctrl_timeout_ms);
if (rc != 0) return rc;
}
// Channel 0 = RTP, anything else (RTCP, etc.) we drop.
// The early frames are real video; keeping them avoids
// a brief blank at PLAY time once the data path picks
// up read_nalu().
if (channel == 0) decode_rtp_h264(rtp);
if (channel == 0) handle_rtp(rtp);
else if (channel == 1) handle_rtcp(rtp);
log_at(LL_TRACE, logger, log_ctx,
"rtsp: buffered $-frame ahead of response "
"(ch=%d len=%u)", channel, static_cast<unsigned>(plen));
continue;
}
int read_rc = read_response(ssl, out, static_cast<int>(prefix));
int read_rc = read_response(ssl, out, ctrl_timeout_ms,
static_cast<int>(prefix));
if (read_rc != 0) return read_rc;
log_at(LL_TRACE, logger, log_ctx,
"rtsp <- %d %s (body=%zu)",
@@ -867,6 +953,7 @@ struct Client::Impl {
void do_teardown_best_effort()
{
if (session_id.empty() || !ssl) return;
if (socket_down.load(std::memory_order_acquire)) return;
// Best-effort fire-and-forget; the response (if any) is
// discarded. We ignore errors here because by the time we
// call this the user is already exiting.
@@ -886,33 +973,173 @@ struct Client::Impl {
void keepalive_main()
{
while (!stop_flag.load(std::memory_order_acquire)) {
std::unique_lock<std::mutex> lk(keepalive_mu);
keepalive_cv.wait_for(lk, std::chrono::seconds(15), [&] {
return stop_flag.load(std::memory_order_acquire);
});
if (stop_flag.load(std::memory_order_acquire)) break;
// GET_PARAMETER with no body is the standard idle ping
// live555 accepts. Build the request inline (instead of
// through send_request) so we keep the Session: header
// even if the data path also wants the io_mu mutex.
std::string auth = compute_authorization("GET_PARAMETER", url_full);
std::string req = std::string("GET_PARAMETER ") + url_full +
" RTSP/1.0\r\n" +
"CSeq: " + std::to_string(cseq++) + "\r\n" +
"Session: " + session_id + "\r\n" +
auth +
"\r\n";
std::lock_guard<std::mutex> io_lk(io_mu);
if (!ssl) break;
if (obn::tls::ssl_write_all(ssl, req.data(), req.size()) != 0) {
log_at(LL_DEBUG, logger, log_ctx, "rtsp: keepalive write failed");
break;
{
std::unique_lock<std::mutex> lk(keepalive_mu);
keepalive_cv.wait_for(lk, std::chrono::milliseconds(kRtspPingMs),
[&] {
return stop_flag.load(std::memory_order_acquire);
});
}
log_at(LL_TRACE, logger, log_ctx, "rtsp: keepalive sent");
if (stop_flag.load(std::memory_order_acquire)) break;
if (!send_rtsp_ping()) break;
}
}
// Called by the reader thread between messages. RTCP lives here
// rather than on the keepalive thread on purpose: the reader wakes
// on every RTP packet, so the cadence stays accurate without a
// second thread writing into the same SSL object.
void maybe_send_rtcp()
{
const auto now = std::chrono::steady_clock::now();
if (now < next_rtcp_due) return;
next_rtcp_due = now + std::chrono::milliseconds(kRtcpIntervalMs);
send_rtcp_receiver_report();
}
// GET_PARAMETER with no body is the standard idle ping live555
// accepts. Built inline (instead of through send_request) so we keep
// the Session: header even while the data path holds io_mu.
bool send_rtsp_ping()
{
if (socket_down.load(std::memory_order_acquire)) return false;
std::string auth = compute_authorization("GET_PARAMETER", url_full);
std::string req = std::string("GET_PARAMETER ") + url_full +
" RTSP/1.0\r\n" +
"CSeq: " + std::to_string(cseq++) + "\r\n" +
"Session: " + session_id + "\r\n" +
auth +
"\r\n";
std::lock_guard<std::mutex> io_lk(io_mu);
if (!ssl) return false;
if (obn::tls::ssl_write_all(ssl, req.data(), req.size()) != 0) {
log_at(LL_DEBUG, logger, log_ctx, "rtsp: keepalive write failed");
return false;
}
log_at(LL_TRACE, logger, log_ctx, "rtsp: keepalive sent");
return true;
}
// Emit a compound RTCP packet (receiver report + SDES/CNAME) on
// interleaved channel 1. This is what tells the printer's RTP sink
// that we are still here; without it the sink eventually stops
// sending and the session dies silently.
bool send_rtcp_receiver_report()
{
if (socket_down.load(std::memory_order_acquire)) return false;
std::vector<std::uint8_t> pkt;
{
std::lock_guard<std::mutex> lk(stats_mu);
// Nothing to report on before the first RTP packet arrives.
if (!seq_initialised || !have_peer_ssrc) return true;
build_receiver_report(&pkt);
}
append_sdes_cname(&pkt);
std::uint8_t frame[4] = {
'$', 1,
static_cast<std::uint8_t>((pkt.size() >> 8) & 0xff),
static_cast<std::uint8_t>(pkt.size() & 0xff),
};
std::lock_guard<std::mutex> io_lk(io_mu);
if (!ssl) return false;
if (obn::tls::ssl_write_all(ssl, frame, sizeof(frame)) != 0 ||
obn::tls::ssl_write_all(ssl, pkt.data(), pkt.size()) != 0) {
log_at(LL_DEBUG, logger, log_ctx, "rtsp: RTCP RR write failed");
return false;
}
log_at(LL_TRACE, logger, log_ctx, "rtsp: RTCP RR sent (%zu bytes)",
pkt.size());
return true;
}
// RFC 3550 section 6.4.2 receiver report with a single report block.
// Caller holds stats_mu.
void build_receiver_report(std::vector<std::uint8_t>* out)
{
const std::uint32_t extended_max = seq_cycles + max_seq;
const std::uint32_t expected = extended_max - base_seq + 1;
std::int64_t lost = static_cast<std::int64_t>(expected) -
static_cast<std::int64_t>(packets_received);
if (lost < 0) lost = 0;
if (lost > 0x7fffff) lost = 0x7fffff;
const std::uint32_t expected_interval = expected - expected_prior;
const std::uint32_t received_interval = packets_received - received_prior;
expected_prior = expected;
received_prior = packets_received;
const std::int64_t lost_interval =
static_cast<std::int64_t>(expected_interval) -
static_cast<std::int64_t>(received_interval);
std::uint8_t fraction = 0;
if (expected_interval > 0 && lost_interval > 0) {
fraction = static_cast<std::uint8_t>(
(lost_interval << 8) / expected_interval);
}
// Delay since the last sender report, in 1/65536 s.
std::uint32_t dlsr = 0;
if (have_sr) {
const auto d = std::chrono::steady_clock::now() - last_sr_at;
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(d)
.count();
dlsr = static_cast<std::uint32_t>((us * 65536) / 1000000);
}
auto push32 = [out](std::uint32_t v) {
out->push_back(static_cast<std::uint8_t>((v >> 24) & 0xff));
out->push_back(static_cast<std::uint8_t>((v >> 16) & 0xff));
out->push_back(static_cast<std::uint8_t>((v >> 8) & 0xff));
out->push_back(static_cast<std::uint8_t>( v & 0xff));
};
out->push_back(0x80 | 1); // V=2, P=0, RC=1
out->push_back(static_cast<std::uint8_t>(kRtcpRr));
out->push_back(0); // length, patched below
out->push_back(7); // 32 bytes = 8 words - 1
push32(our_ssrc);
push32(peer_ssrc);
out->push_back(fraction);
out->push_back(static_cast<std::uint8_t>((lost >> 16) & 0xff));
out->push_back(static_cast<std::uint8_t>((lost >> 8) & 0xff));
out->push_back(static_cast<std::uint8_t>( lost & 0xff));
push32(extended_max);
push32(static_cast<std::uint32_t>(jitter));
push32(have_sr ? last_sr_middle32 : 0);
push32(dlsr);
}
// Minimal SDES chunk carrying just a CNAME, which report receivers
// use to correlate our RR with our RTP source.
void append_sdes_cname(std::vector<std::uint8_t>* out)
{
char cname[32];
const int cn = std::snprintf(cname, sizeof(cname), "obn-%08x",
static_cast<unsigned>(our_ssrc));
const std::size_t cname_len = (cn > 0) ? static_cast<std::size_t>(cn) : 0;
const std::size_t start = out->size();
out->push_back(0x80 | 1); // V=2, P=0, SC=1
out->push_back(static_cast<std::uint8_t>(kRtcpSdes));
out->push_back(0); // length, patched below
out->push_back(0);
out->push_back(static_cast<std::uint8_t>((our_ssrc >> 24) & 0xff));
out->push_back(static_cast<std::uint8_t>((our_ssrc >> 16) & 0xff));
out->push_back(static_cast<std::uint8_t>((our_ssrc >> 8) & 0xff));
out->push_back(static_cast<std::uint8_t>( our_ssrc & 0xff));
out->push_back(1); // CNAME
out->push_back(static_cast<std::uint8_t>(cname_len));
out->insert(out->end(), cname, cname + cname_len);
out->push_back(0); // end of item list
while ((out->size() - start) % 4 != 0) out->push_back(0);
const std::size_t words = (out->size() - start) / 4 - 1;
(*out)[start + 2] = static_cast<std::uint8_t>((words >> 8) & 0xff);
(*out)[start + 3] = static_cast<std::uint8_t>(words & 0xff);
}
// ----- demux -----
// Read the next chunk on the wire, classify it, and either:
@@ -923,12 +1150,12 @@ struct Client::Impl {
int pull_one_message()
{
std::uint8_t prefix;
int rc = obn::tls::ssl_read_full(ssl, &prefix, 1);
int rc = obn::tls::ssl_read_full_timeout(ssl, &prefix, 1, data_timeout_ms);
if (rc != 0) return rc;
if (prefix == '$') {
// Interleaved RTP/RTCP frame.
std::uint8_t hdr[3];
rc = obn::tls::ssl_read_full(ssl, hdr, 3);
rc = obn::tls::ssl_read_full_timeout(ssl, hdr, 3, data_timeout_ms);
if (rc != 0) return rc;
int channel = hdr[0];
std::uint16_t plen = static_cast<std::uint16_t>(
@@ -938,12 +1165,17 @@ struct Client::Impl {
return -1;
}
std::vector<std::uint8_t> rtp(plen);
rc = obn::tls::ssl_read_full(ssl, rtp.data(), plen);
rc = obn::tls::ssl_read_full_timeout(ssl, rtp.data(), plen,
data_timeout_ms);
if (rc != 0) return rc;
// Channel 0 = RTP, channel 1 = RTCP. Bambu only sends 0
// for video, but we filter defensively.
if (channel != 0) return 0;
decode_rtp_h264(rtp);
// Channel 0 = RTP, channel 1 = RTCP (sender reports we
// answer, plus BYE when the server closes the session).
if (channel == 0) {
handle_rtp(rtp);
} else if (channel == 1) {
handle_rtcp(rtp);
if (peer_bye) return 1;
}
return 0;
}
if (prefix == 'R') {
@@ -951,7 +1183,7 @@ struct Client::Impl {
// Re-prepend the 'R' we already consumed and parse normally.
std::string status = "R";
std::string rest;
rc = obn::tls::ssl_read_line(ssl, &rest);
rc = obn::tls::ssl_read_line_timeout(ssl, &rest, data_timeout_ms);
if (rc != 0) return rc;
status += rest;
(void)status;
@@ -959,7 +1191,7 @@ struct Client::Impl {
std::size_t content_length = 0;
for (;;) {
std::string line;
rc = obn::tls::ssl_read_line(ssl, &line);
rc = obn::tls::ssl_read_line_timeout(ssl, &line, data_timeout_ms);
if (rc != 0) return rc;
if (line.empty()) break;
auto colon = line.find(':');
@@ -976,7 +1208,9 @@ struct Client::Impl {
}
if (content_length > 0) {
std::vector<std::uint8_t> body(content_length);
rc = obn::tls::ssl_read_full(ssl, body.data(), content_length);
rc = obn::tls::ssl_read_full_timeout(ssl, body.data(),
content_length,
data_timeout_ms);
if (rc != 0) return rc;
}
log_at(LL_TRACE, logger, log_ctx, "rtsp: drained stray response");
@@ -993,6 +1227,104 @@ struct Client::Impl {
return -1;
}
// Account one received RTP packet, then depacketise it. The stats
// feed the RTCP receiver reports the keepalive thread sends.
void handle_rtp(const std::vector<std::uint8_t>& rtp)
{
if (rtp.size() >= 12) {
const std::uint16_t seq = static_cast<std::uint16_t>(
(std::uint16_t(rtp[2]) << 8) | rtp[3]);
const std::uint32_t ts = (std::uint32_t(rtp[4]) << 24)
| (std::uint32_t(rtp[5]) << 16)
| (std::uint32_t(rtp[6]) << 8)
| std::uint32_t(rtp[7]);
const std::uint32_t ssrc = (std::uint32_t(rtp[8]) << 24)
| (std::uint32_t(rtp[9]) << 16)
| (std::uint32_t(rtp[10]) << 8)
| std::uint32_t(rtp[11]);
update_rtp_stats(seq, ts, ssrc);
}
decode_rtp_h264(rtp);
}
void update_rtp_stats(std::uint16_t seq, std::uint32_t rtp_ts,
std::uint32_t ssrc)
{
std::lock_guard<std::mutex> lk(stats_mu);
peer_ssrc = ssrc;
have_peer_ssrc = true;
if (!seq_initialised) {
seq_initialised = true;
base_seq = seq;
max_seq = seq;
seq_cycles = 0;
packets_received = 0;
expected_prior = 0;
received_prior = 0;
} else {
// RFC 3550 A.1: a small forward delta advances the highest
// sequence number and wraps the cycle counter; anything else
// is a duplicate or a reorder and only counts as received.
const std::uint16_t delta = static_cast<std::uint16_t>(seq - max_seq);
if (delta < 0x8000) {
if (seq < max_seq) seq_cycles += 0x10000;
max_seq = seq;
}
}
++packets_received;
// Interarrival jitter (RFC 3550 A.8) in RTP timestamp units.
const auto now = std::chrono::steady_clock::now();
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
now - stream_t0).count();
const std::int64_t arrival =
(us * static_cast<std::int64_t>(track.clock_rate)) / 1000000;
const std::int64_t transit = arrival - static_cast<std::int64_t>(rtp_ts);
if (have_transit) {
std::int64_t d = transit - last_transit;
if (d < 0) d = -d;
jitter += (static_cast<double>(d) - jitter) / 16.0;
}
last_transit = transit;
have_transit = true;
}
// Walk a compound RTCP packet: remember the newest sender report so
// the next receiver report can echo it, and notice a BYE.
void handle_rtcp(const std::vector<std::uint8_t>& pkt)
{
std::size_t off = 0;
while (off + 4 <= pkt.size()) {
if ((pkt[off] >> 6) != 2) return; // not RTCP v2
const int pt = pkt[off + 1];
const std::size_t words = (std::size_t(pkt[off + 2]) << 8) | pkt[off + 3];
const std::size_t len = (words + 1) * 4;
if (off + len > pkt.size()) return;
if (pt == kRtcpSr && len >= 20) {
// NTP timestamp lives at +8; the report echoes its
// middle 32 bits (low 16 of seconds, high 16 of fraction).
const std::uint32_t msw = (std::uint32_t(pkt[off + 8]) << 24)
| (std::uint32_t(pkt[off + 9]) << 16)
| (std::uint32_t(pkt[off + 10]) << 8)
| std::uint32_t(pkt[off + 11]);
const std::uint32_t lsw = (std::uint32_t(pkt[off + 12]) << 24)
| (std::uint32_t(pkt[off + 13]) << 16)
| (std::uint32_t(pkt[off + 14]) << 8)
| std::uint32_t(pkt[off + 15]);
std::lock_guard<std::mutex> lk(stats_mu);
last_sr_middle32 = (msw << 16) | (lsw >> 16);
last_sr_at = std::chrono::steady_clock::now();
have_sr = true;
} else if (pt == kRtcpBye) {
log_fmt(logger, log_ctx, "rtsp: server sent RTCP BYE");
peer_bye = true;
return;
}
off += len;
}
}
void decode_rtp_h264(const std::vector<std::uint8_t>& rtp)
{
if (rtp.size() < 12) return;
@@ -1115,6 +1447,9 @@ int Client::start(const Url& url, int connect_timeout_ms)
"rtsp: dialing %s://%s:%d",
url.tls ? "rtsps" : "rtsp", url.host.c_str(), url.port);
I.ctrl_timeout_ms = (connect_timeout_ms > 0) ? connect_timeout_ms : 5000;
I.data_timeout_ms = data_read_timeout_ms();
if (url.tls) {
if (obn::tls::dial_tls(url.host, url.port, connect_timeout_ms,
&I.fd, &I.ssl,
@@ -1125,6 +1460,14 @@ int Client::start(const Url& url, int connect_timeout_ms)
}
log_fmt(I.logger, I.log_ctx, "rtsp: TLS established (cipher=%s)",
SSL_get_cipher(I.ssl));
// dial_tls leaves the socket fully blocking. The read helpers bound
// their own receives, but nothing bounds a *write*: the keepalive
// and RTCP senders would wedge on a printer that stopped draining
// its receive window. Give both directions the tighter of the two
// budgets, which also spares the read path a pair of setsockopt
// calls per read since it then finds a timeout it can live with.
obn::tls::set_socket_io_timeout(
I.fd, std::min(I.ctrl_timeout_ms, I.data_timeout_ms));
} else {
I.fd = obn::tls::dial(url.host, url.port, connect_timeout_ms);
if (!obn::os::socket_valid(I.fd)) {
@@ -1162,12 +1505,18 @@ int Client::start(const Url& url, int connect_timeout_ms)
impl_->track.rtp_pt, impl_->track.clock_rate,
impl_->track.sps.size(), impl_->track.pps.size(),
I.url_setup.c_str());
if (impl_->track.width > 0) {
log_fmt(I.logger, I.log_ctx, "rtsp: SPS geometry %dx%d @ %d fps",
impl_->track.width, impl_->track.height, impl_->track.fps);
}
}
if (I.do_setup(url) != 0) goto fail;
if (I.do_play(url) != 0) goto fail;
log_fmt(I.logger, I.log_ctx, "rtsp: PLAY ok, session=%s", I.session_id.c_str());
I.keepalive = std::thread(&Impl::keepalive_main, impl_.get());
I.stream_t0 = std::chrono::steady_clock::now();
I.next_rtcp_due = I.stream_t0 + std::chrono::milliseconds(kRtcpIntervalMs);
I.keepalive = std::thread(&Impl::keepalive_main, impl_.get());
return 0;
fail:
@@ -1184,6 +1533,7 @@ int Client::read_nalu(Nalu* out)
auto& I = *impl_;
while (I.nalu_queue.empty()) {
if (I.stop_flag.load(std::memory_order_acquire) || !I.ssl) return -1;
I.maybe_send_rtcp();
int rc = I.pull_one_message();
if (rc != 0) return rc;
}
@@ -1198,9 +1548,13 @@ void Client::stop()
auto& I = *impl_;
if (I.stop_flag.exchange(true)) return;
I.keepalive_cv.notify_all();
if (obn::os::socket_valid(I.fd)) obn::os::shutdown_both(I.fd);
if (I.keepalive.joinable()) I.keepalive.join();
// TEARDOWN is worth sending only while the connection is still up;
// after a cancel() it is a write to a shut-down socket. Closing the
// TCP connection ends the session on the server either way.
I.do_teardown_best_effort();
I.socket_down.store(true, std::memory_order_release);
if (obn::os::socket_valid(I.fd)) obn::os::shutdown_both(I.fd);
std::lock_guard<std::mutex> lk(I.io_mu);
obn::tls::close_tls(&I.fd, &I.ssl);
}
@@ -1209,6 +1563,7 @@ void Client::cancel()
{
if (!impl_) return;
auto& I = *impl_;
I.socket_down.store(true, std::memory_order_release);
if (obn::os::socket_valid(I.fd)) obn::os::shutdown_both(I.fd);
I.keepalive_cv.notify_all();
}
stubs/rtsp_client.hpp
+11 -2
View File
@@ -30,8 +30,8 @@
// read_nalu() in a loop and another thread calling stop() exactly
// once. cancel() is the only call that may be made from any thread:
// it shuts the socket down to break the producer out of an in-flight
// SSL_read. The 15-second GET_PARAMETER keepalive runs on its own
// internal thread that the client owns.
// SSL_read. The RTSP keepalive and the RTCP receiver reports run on
// one internal thread that the client owns.
#pragma once
#include <cstdint>
@@ -76,6 +76,15 @@ struct H264Track {
// emit NAL units originating from this PT; others (RTCP, audio)
// are dropped before reaching read_nalu().
int rtp_pt = 96;
// Geometry and frame rate decoded from the SDP's SPS, cropping
// applied. Zero when the SDP carried no SPS or it did not parse,
// in which case the caller keeps its own default.
int width = 0;
int height = 0;
// Rounded frames per second from the SPS VUI timing info; 0 when
// the SPS omits it (VUI timing is optional).
int fps = 0;
};
// One NAL unit handed back to the caller. `data` is *raw NAL bytes*:
stubs/rtsp_passthrough.cpp
+56 -1
View File
@@ -6,6 +6,7 @@
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <mutex>
@@ -28,6 +29,21 @@ constexpr int kFlagSync = 1;
// units gives ~250 ms of headroom which is plenty.
constexpr std::size_t kMaxReadyQueue = 8;
// How long try_pull() keeps answering "would block" before declaring
// the stream dead. Deliberately longer than the RTSP client's own read
// deadline so the more precise error from the reader thread wins when
// both would fire. OBN_RTSP_STALL_TIMEOUT_MS overrides it.
constexpr int kStallTimeoutMs = 8000;
int stall_timeout_ms()
{
if (const char* e = std::getenv("OBN_RTSP_STALL_TIMEOUT_MS")) {
const int v = std::atoi(e);
if (v > 0) return v;
}
return kStallTimeoutMs;
}
// One Annex-B encoded access unit ready to be handed to the caller.
struct Sample {
std::vector<std::uint8_t> bytes;
@@ -61,6 +77,13 @@ struct Passthrough::Impl {
std::condition_variable q_cv;
std::deque<Sample> ready;
// Watchdog state, guarded by q_mu. `last_progress` moves forward on
// every produced access unit; `stalled` latches once try_pull has
// reported the stall so the error is stable across calls.
std::chrono::steady_clock::time_point last_progress;
bool stalled = false;
int stall_ms = kStallTimeoutMs;
Sample current; // borrowed by last try_pull
std::chrono::steady_clock::time_point t0;
@@ -109,6 +132,12 @@ int Passthrough::start(const std::string& host,
tr.sps.size(), tr.pps.size(), tr.rtp_pt);
I.t0 = std::chrono::steady_clock::now();
{
std::lock_guard<std::mutex> lk(I.q_mu);
I.last_progress = I.t0;
I.stalled = false;
I.stall_ms = stall_timeout_ms();
}
I.stop_flag.store(false, std::memory_order_release);
I.worker = std::thread([this] {
auto& I = *impl_;
@@ -155,6 +184,7 @@ int Passthrough::start(const std::string& host,
I.ready.pop_front();
}
I.ready.emplace_back(std::move(s));
I.last_progress = now;
lk.unlock();
I.q_cv.notify_all();
};
@@ -190,6 +220,7 @@ int Passthrough::start(const std::string& host,
marker.flags = (rc == 1 || stopping)
? 0x100 /*EOS*/ : 0x200 /*ERR*/;
I.ready.emplace_back(std::move(marker));
I.last_progress = std::chrono::steady_clock::now();
lk.unlock();
I.q_cv.notify_all();
break;
@@ -235,6 +266,16 @@ int Passthrough::start(const std::string& host,
return 0;
}
Passthrough::StreamParams Passthrough::params() const
{
const auto& tr = impl_->client.track();
StreamParams p;
p.width = tr.width;
p.height = tr.height;
p.fps = tr.fps;
return p;
}
Passthrough::PullResult Passthrough::try_pull(const std::uint8_t** out_buf,
std::size_t* out_size,
std::uint64_t* out_dt_100ns,
@@ -242,7 +283,21 @@ Passthrough::PullResult Passthrough::try_pull(const std::uint8_t** out_buf,
{
auto& I = *impl_;
std::unique_lock<std::mutex> lk(I.q_mu);
if (I.ready.empty()) return Pull_WouldBlock;
if (I.ready.empty()) {
if (I.stalled) return Pull_Error;
const auto idle = std::chrono::steady_clock::now() - I.last_progress;
if (std::chrono::duration_cast<std::chrono::milliseconds>(idle).count() >=
I.stall_ms) {
I.stalled = true;
lk.unlock();
log_fmt(I.logger, I.log_ctx,
"rtsp_passthrough: no video for %d ms -- reporting the "
"stream as dead so the pipeline can be torn down",
I.stall_ms);
return Pull_Error;
}
return Pull_WouldBlock;
}
I.current = std::move(I.ready.front());
I.ready.pop_front();
stubs/rtsp_passthrough.hpp
+17
View File
@@ -49,6 +49,13 @@ public:
Pull_Error = -1,
};
// Geometry the SDP's SPS advertised, zeroed when it did not parse.
struct StreamParams {
int width = 0;
int height = 0;
int fps = 0;
};
Passthrough(obn::source::Logger logger, void* log_ctx);
~Passthrough();
@@ -64,6 +71,9 @@ public:
const std::string& device = {},
int connect_timeout_ms = 5000);
// Valid after a successful start().
StreamParams params() const;
// Non-blocking pop of the next ready Annex-B sample. The caller
// gets a borrowed pointer; it remains valid until the next
// try_pull() call (or stop()). dt_100ns is wall-clock since
@@ -71,6 +81,13 @@ public:
// gstbambusrc's `decode_time * 100ULL` does the right thing).
// flags carries Bambu_SampleFlag::f_sync (1) on access units
// containing an IDR slice.
//
// Never answers Pull_WouldBlock forever: once no access unit has
// been produced for the stall timeout, every call returns
// Pull_Error. gstbambusrc polls "would block" in a usleep loop
// while holding the pad's stream lock and has no unlock() handler,
// so an endlessly blocked source deadlocks the slicer's teardown
// (issue #54) -- reporting the dead stream is what breaks the loop.
PullResult try_pull(const std::uint8_t** out_buf,
std::size_t* out_size,
std::uint64_t* out_dt_100ns,
tests/h264_avcc_test.cpp
+43
View File
@@ -147,6 +147,49 @@ int main()
CHECK(f.contains_idr);
}
// 7. parse_sps_geometry against real x264 output. Both fixtures are
// the SPS NAL as it appears on the wire; ffmpeg reports 1168x720
// @ 24.67 fps and 1280x720 @ 30 fps for them respectively, and
// the first one exercises the crop path (80 macroblocks = 1280
// coded pixels cropped down to 1168) plus the High-profile
// chroma_format_idc block.
{
const uint8_t sps_high_1168x720[] = {
0x67, 0x64, 0x00, 0x1f, 0xac, 0xd9, 0x40, 0x49, 0x05, 0xbb,
0x01, 0x10, 0x00, 0x00, 0x03, 0x00, 0x30, 0x00, 0x00, 0x09,
0x40, 0xf1, 0x83, 0x19, 0x60,
};
obn::h264::SpsGeometry g;
CHECK(obn::h264::parse_sps_geometry(sps_high_1168x720,
sizeof(sps_high_1168x720), &g));
CHECK(g.width == 1168);
CHECK(g.height == 720);
CHECK(g.fps == 25); // 24.67 fps, rounded
}
{
const uint8_t sps_baseline_1280x720[] = {
0x67, 0x42, 0xc0, 0x1f, 0xd9, 0x00, 0x50, 0x05, 0xbb, 0x01,
0x10, 0x00, 0x00, 0x03, 0x00, 0x10, 0x00, 0x00, 0x03, 0x03,
0xc0, 0xf1, 0x83, 0x24, 0x80,
};
obn::h264::SpsGeometry g;
CHECK(obn::h264::parse_sps_geometry(sps_baseline_1280x720,
sizeof(sps_baseline_1280x720), &g));
CHECK(g.width == 1280);
CHECK(g.height == 720);
CHECK(g.fps == 30);
}
// 8. Truncated and empty SPS inputs must fail rather than read past
// the buffer or invent a geometry.
{
const uint8_t truncated[] = { 0x67, 0x64, 0x00 };
obn::h264::SpsGeometry g;
CHECK(!obn::h264::parse_sps_geometry(truncated, sizeof(truncated), &g));
CHECK(!obn::h264::parse_sps_geometry(nullptr, 0, &g));
CHECK(g.width == 0 && g.height == 0 && g.fps == 0);
}
if (g_fail) {
std::printf("h264_avcc_test FAILED (%d)\n", g_fail);
return 1;