GCC Code Coverage Report

Directory: ./
File: mediaoutput/OpusRTPManager.cpp
Date: 2026-10-29 08:11:58
Exec Total Coverage
Lines: 449 603 74.5%
Functions: 27 30 90.0%
Branches: 486 1137 42.7%
Line Branch Exec Source
1 /*
2 * This file is part of the Falcon Player (FPP) and is Copyright (C)
3 * 2013-2025 by the Falcon Player Developers.
4 *
5 * The Falcon Player (FPP) is free software, and is covered under
6 * multiple Open Source licenses. Please see the included 'LICENSES'
7 * file for descriptions of what files are covered by each license.
8 *
9 * This source file is covered under the LGPL v2.1 as described in the
10 * included LICENSE.LGPL file.
11 */
12
13 #include "fpp-pch.h"
14
15 #include "fpp-json.h"
16 #include "fpphttp.h" // drogon/HTTP helpers used here; no longer pulled transitively (see fpphttp_types.h)
17
18 #include "Warnings.h" // WarningHolder -- needed directly for NOPCH builds
19
20 #include "OpusRTPManager.h"
21 #include "GStreamerOut.h"
22 #include "PipeWireGraphConfig.h"
23
24 #ifdef HAS_OPUS_RTP_GSTREAMER
25
26 #include <gst/gst.h>
27
28 #include <arpa/inet.h>
29 #include <cctype>
30 #include <ifaddrs.h>
31 #include <net/if.h>
32 #include <netinet/in.h>
33 #include <sys/socket.h>
34
35 #include <algorithm>
36 #include <chrono>
37 #include <cstring>
38 #include <sstream>
39
40 #include "common.h"
41 #include "log.h"
42 #include "mediaoutput.h"
43 #include "settings.h"
44
45 // ──────────────────────────────────────────────────────────────────────────────
46 // Singleton instance
47 // ──────────────────────────────────────────────────────────────────────────────
48 static OpusRTPManager s_opusRTPManager;
49 OpusRTPManager& OpusRTPManager::INSTANCE = s_opusRTPManager;
50
51 // Set a send pipeline's master volume element (see SetMasterVolume()).
52 3 static void SetPipelineMasterVolume(GstElement* pipeline, int volume) {
53 3 GstElement* vol = gst_bin_get_by_name(GST_BIN(pipeline), "mastervol");
54
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 if (vol) {
55 3 g_object_set(vol, "volume", masterVolumeGain(volume), nullptr);
56 3 gst_object_unref(vol);
57 }
58 3 }
59
60 2346 OpusRTPManager::OpusRTPManager() {
61
2/4
✓ Branch 0 taken 2346 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2346 times.
✗ Branch 3 not taken.
 2346 m_configPath = getFPPMediaDir("/config/pipewire-opus-rtp-instances.json");
62 2346 }
63
64 916 OpusRTPManager::~OpusRTPManager() {
65 916 Shutdown();
66 916 }
67
68 // ──────────────────────────────────────────────────────────────────────────────
69 // Lifecycle
70 // ──────────────────────────────────────────────────────────────────────────────
71 3761 bool OpusRTPManager::Init() {
72
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3761 times.
 3761 if (m_initialized.load()) {
73 ✗ return true;
74 }
75
76 // Opus RTP uses pipewiresrc/pipewiresink which require the full PipeWire graph.
77 // Only proceed when MediaBackend is "pipewire" (the advanced PipeWire mode).
78 // - ALSA / no backend: PipeWire daemon is not running; gst_parse_launch with
79 // pipewiresrc crashes in gst_value_deserialize.
80 // - pipewire-simple: the graph lacks the node connections needed for audio
81 // format negotiation, causing the state change to block indefinitely.
82
2/4
✓ Branch 0 taken 3761 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3761 times.
✗ Branch 3 not taken.
 3761 std::string mediaBackend = toLowerCopy(getSetting("MediaBackend"));
83
8/10
✓ Branch 0 taken 3761 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3749 times.
✓ Branch 3 taken 12 times.
✓ Branch 4 taken 3749 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3610 times.
✓ Branch 7 taken 139 times.
✓ Branch 8 taken 3622 times.
✓ Branch 9 taken 139 times.
 3761 if (!isPipeWireBackend() || mediaBackend != "pipewire") {
84
1/2
✓ Branch 0 taken 3622 times.
✗ Branch 1 not taken.
 3622 LogDebug(VB_MEDIAOUT, "OpusRTPManager: MediaBackend='%s' (need 'pipewire'), skipping init\n",
85 mediaBackend.c_str());
86 3622 return true;
87 }
88
89
3/4
✓ Branch 0 taken 139 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 126 times.
✓ Branch 3 taken 13 times.
 139 if (!FileExists(m_configPath)) {
90
1/2
✓ Branch 0 taken 126 times.
✗ Branch 1 not taken.
 126 LogDebug(VB_MEDIAOUT, "OpusRTPManager: No config file at %s, skipping init\n",
91 m_configPath.c_str());
92 126 return true;
93 }
94
95 // Shared with the playback path and the video managers so gst_init()
96 // happens exactly once, under one lock, whichever subsystem gets here
97 // first.
98
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 GStreamerOutput::EnsureGStreamerInit();
99
100 // Set PipeWire env vars
101 13 setenv("PIPEWIRE_RUNTIME_DIR", "/run/pipewire-fpp", 0);
102 13 setenv("XDG_RUNTIME_DIR", "/run/pipewire-fpp", 0);
103 13 setenv("PULSE_RUNTIME_PATH", "/run/pipewire-fpp/pulse", 0);
104
105 13 m_initialized.store(true);
106
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 LogInfo(VB_MEDIAOUT, "OpusRTPManager: Initialized\n");
107 13 return true;
108 3761 }
109
110 2325 void OpusRTPManager::Shutdown() {
111
2/2
✓ Branch 0 taken 2312 times.
✓ Branch 1 taken 13 times.
 2325 if (!m_initialized.load()) {
112 2312 return;
113 }
114
115
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 LogInfo(VB_MEDIAOUT, "OpusRTPManager: Shutting down\n");
116
117 // Nothing may apply from here on, including a watchdog rebuild that is
118 // already scheduled: ApplyConfig() would re-Init() and bring back the
119 // pipelines this is about to tear down.
120 13 m_stopping.store(true);
121 13 m_applyGeneration++;
122 13 m_initialized.store(false);
123
124 // Join the rebuild thread BEFORE taking m_applyMutex: that thread calls
125 // ApplyConfig(), which itself takes the lock -- holding it across the
126 // join would deadlock. Taken out under m_rebuildMutex: the watchdog
127 // thread assigns it.
128 13 std::thread rebuild;
129 {
130
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 std::lock_guard<std::mutex> rl(m_rebuildMutex);
131 13 rebuild = std::move(m_rebuildThread);
132 13 }
133
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 12 times.
 13 if (rebuild.joinable()) {
134
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 rebuild.join();
135 }
136
137 // Only now take the apply lock -- it blocks until any in-flight
138 // ApplyConfig() (e.g. from a command thread) has finished.
139
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 std::lock_guard<std::mutex> applyLock(m_applyMutex);
140
141 13 m_watchdogRunning.store(false);
142
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 7 times.
 13 if (m_watchdogThread.joinable()) {
143
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 m_watchdogThread.join();
144 }
145
146
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 StopAllPipelines();
147
148 13 m_active.store(false);
149
1/2
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
 13 LogInfo(VB_MEDIAOUT, "OpusRTPManager: Shutdown complete\n");
150 13 }
151
152 // ──────────────────────────────────────────────────────────────────────────────
153 // Config loading
154 // ──────────────────────────────────────────────────────────────────────────────
155 584 OpusRTPConfig OpusRTPManager::GetConfigSnapshot() {
156
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 std::lock_guard<std::mutex> lock(m_configMutex);
157
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 1168 return m_config;
158 584 }
159
160
161 // Values that go into a gst_parse_launch() description unquoted. Anything
162 // outside these characters is refused at load: a space or '!' in a
163 // destination address re-parses as extra pipeline elements (a udpsrc feeding
164 // a filesink is a root file write), and no real address or interface name
165 // needs them.
166 96 static bool IsPipelineSafeHost(const std::string& host) {
167
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 96 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 96 times.
 96 if (host.empty() || host.size() > 253) {
168 ✗ return false;
169 }
170
2/2
✓ Branch 0 taken 864 times.
✓ Branch 1 taken 96 times.
 960 for (unsigned char c : host) {
171
3/8
✓ Branch 0 taken 288 times.
✓ Branch 1 taken 576 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 288 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
 864 if (!std::isalnum(c) && c != '.' && c != '-' && c != ':') {
172 ✗ return false;
173 }
174 }
175 96 return true;
176 }
177 96 static bool IsPipelineSafeInterface(const std::string& iface) {
178
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 96 times.
 96 if (iface.size() > 15) { // IFNAMSIZ - 1
179 ✗ return false;
180 }
181
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 96 times.
 96 for (unsigned char c : iface) {
182 ✗ if (!std::isalnum(c) && c != '.' && c != '-' && c != '_' && c != '@') {
183 ✗ return false;
184 }
185 }
186 96 return true;
187 }
188
189 // The pipeline-start warnings. Kept as constants because RemoveWarning()
190 // matches the exact text: an apply where every stream of a kind started
191 // retracts that kind's warning, and an apply with nothing to run or Cleanup()
192 // retracts both -- before this, nothing ever did, and the banner outlived the
193 // fix until fppd restarted.
194 static const char* const OPUS_WARNING_SEND_FAILED = "Opus RTP: audio send stream failed to start";
195 static const char* const OPUS_WARNING_RECV_FAILED = "Opus RTP: audio receive stream failed to start";
196 // Raised when an apply can't read the config, so nothing runs; the next apply
197 // (any outcome) or Cleanup() retracts it.
198 static const char* const OPUS_WARNING_CONFIG_UNREADABLE = "Opus RTP: pipewire-opus-rtp-instances.json could not be read, so no stream is running";
199 73 static void ClearOpusPipelineWarnings(bool send, bool recv) {
200
2/2
✓ Branch 0 taken 45 times.
✓ Branch 1 taken 28 times.
 73 if (send) {
201
3/6
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 45 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 45 times.
✗ Branch 5 not taken.
 180 WarningHolder::RemoveWarning(44, OPUS_WARNING_SEND_FAILED);
202 }
203
2/2
✓ Branch 0 taken 65 times.
✓ Branch 1 taken 8 times.
 73 if (recv) {
204
3/6
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 65 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 65 times.
✗ Branch 5 not taken.
 260 WarningHolder::RemoveWarning(44, OPUS_WARNING_RECV_FAILED);
205 }
206 73 }
207
208 71 bool OpusRTPManager::LoadConfig() {
209
1/2
✓ Branch 0 taken 71 times.
✗ Branch 1 not taken.
 71 Json::Value root;
210
3/4
✓ Branch 0 taken 71 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 69 times.
 71 if (!LoadJsonFromFile(m_configPath, root, JsonRoot::Object)) {
211
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 LogWarn(VB_MEDIAOUT, "OpusRTPManager: Failed to load config from %s\n",
212 m_configPath.c_str());
213 2 return false;
214 }
215
216 // Parse into a local config and publish it in one locked swap at the end.
217 // Filling m_config in place would let a status query on another thread
218 // iterate the vector while push_back() reallocates it.
219 69 OpusRTPConfig cfg;
220
3/4
✓ Branch 0 taken 12 times.
✓ Branch 1 taken 57 times.
✓ Branch 2 taken 12 times.
✗ Branch 3 not taken.
 69 static const OpusRTPConfig kDefault;
221 // Values of the wrong JSON type threw from here to std::terminate; such
222 // a field is left at its default, such an instance skipped.
223 try {
224 69 cfg.requireGroupSource =
225
3/6
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 69 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 69 times.
✗ Branch 5 not taken.
 69 root.get("requireGroupSource", kDefault.requireGroupSource).asBool();
226 ✗ } catch (const std::exception& e) {
227 ✗ WarningHolder::InvalidConfigEntry("pipewire-opus-rtp-instances.json", e.what());
228 ✗ }
229
230
6/12
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 69 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 69 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 69 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 69 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 69 times.
✗ Branch 11 not taken.
 69 if (root.isMember("instances") && root["instances"].isArray()) {
231
8/14
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 69 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 69 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 97 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 97 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 166 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 97 times.
✓ Branch 13 taken 69 times.
 166 for (const auto& instJson : root["instances"]) {
232
1/2
✓ Branch 0 taken 97 times.
✗ Branch 1 not taken.
 97 OpusRTPInstance inst;
233 try {
234
4/6
✓ Branch 0 taken 97 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 97 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✓ Branch 5 taken 1 times.
 99 inst.id = instJson.get("id", 0).asInt();
235
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.name = instJson.get("name", "Opus RTP").asString();
236
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.enabled = instJson.get("enabled", true).asBool();
237
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.mode = instJson.get("mode", "send").asString();
238
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.destIP = instJson.get("destIP", OpusRTP::DEFAULT_DEST_IP).asString();
239
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.port = instJson.get("port", OpusRTP::DEFAULT_PORT).asInt();
240
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.channels = instJson.get("channels", OpusRTP::DEFAULT_CHANNELS).asInt();
241
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.interface = instJson.get("interface", "").asString();
242
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.latency = instJson.get("latency", OpusRTP::DEFAULT_LATENCY_MS).asInt();
243
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.bitrate = instJson.get("bitrate", OpusRTP::DEFAULT_BITRATE).asInt();
244
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.dtx = instJson.get("dtx", false).asBool();
245
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.fec = instJson.get("fec", true).asBool();
246
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 96 times.
✗ Branch 5 not taken.
 96 inst.packetLoss = instJson.get("packetLoss", OpusRTP::DEFAULT_PACKET_LOSS).asInt();
247
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 } catch (const std::exception& e) {
248
3/6
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✗ Branch 5 not taken.
 3 WarningHolder::InvalidConfigEntry("pipewire-opus-rtp-instances.json", e.what());
249 1 continue;
250 1 }
251
252
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 96 times.
 96 if (!OpusRTP::IsValidBitrate(inst.bitrate)) {
253 ✗ inst.bitrate = OpusRTP::DEFAULT_BITRATE;
254 }
255
256
3/6
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 96 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 96 times.
 96 if (!IsPipelineSafeHost(inst.destIP) || !IsPipelineSafeInterface(inst.interface)) {
257 ✗ LogErr(VB_MEDIAOUT, "OpusRTPManager: instance '%s' has an invalid address or interface, skipping it\n",
258 inst.name.c_str());
259 ✗ WarningHolder::AddWarningTimeout(300, 44, "Opus RTP: stream '" + inst.name +
260 ✗ "' has an invalid address or interface and was not started");
261 ✗ continue;
262 }
263
264
1/2
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
 96 cfg.instances.push_back(inst);
265
2/2
✓ Branch 0 taken 96 times.
✓ Branch 1 taken 1 times.
 97 }
266 }
267
268
1/2
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
 69 LogInfo(VB_MEDIAOUT, "OpusRTPManager: Loaded config with %d instances\n",
269 (int)cfg.instances.size());
270
271 {
272
1/2
✓ Branch 0 taken 69 times.
✗ Branch 1 not taken.
 69 std::lock_guard<std::mutex> lock(m_configMutex);
273 69 m_config = std::move(cfg);
274 69 }
275 69 return true;
276 71 }
277
278 // ──────────────────────────────────────────────────────────────────────────────
279 // ApplyConfig
280 // ──────────────────────────────────────────────────────────────────────────────
281 1945 bool OpusRTPManager::ApplyConfig() {
282 // Serialize against concurrent ApplyConfig()/Shutdown()/Cleanup() calls -
283 // see m_applyMutex. Without this, two callers can both get past the
284 // watchdog join below and both reach the std::thread assignment at the
285 // end, where assigning over a joinable thread calls std::terminate().
286
1/2
✓ Branch 0 taken 1945 times.
✗ Branch 1 not taken.
 1945 std::lock_guard<std::mutex> applyLock(m_applyMutex);
287
1/2
✓ Branch 0 taken 1945 times.
✗ Branch 1 not taken.
 3890 return ApplyConfigLocked();
288 1945 }
289
290 1946 bool OpusRTPManager::ApplyConfigLocked() {
291
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1946 times.
 1946 if (m_stopping.load()) {
292 ✗ return false;
293 }
294 // Any rebuild scheduled before this apply is now redundant.
295 1946 m_applyGeneration++;
296
297
2/2
✓ Branch 0 taken 1874 times.
✓ Branch 1 taken 72 times.
 1946 if (!m_initialized.load()) {
298
2/4
✓ Branch 0 taken 1874 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1874 times.
 1874 if (!Init()) {
299 ✗ return false;
300 }
301 // Init() may return true but skip initialization (e.g. wrong backend).
302 // If m_initialized is still false, there's nothing to do.
303
1/2
✓ Branch 0 taken 1874 times.
✗ Branch 1 not taken.
 1874 if (!m_initialized.load()) {
304 1874 return true;
305 }
306 }
307
308 // Stop existing watchdog and pipelines
309 72 m_watchdogRunning.store(false);
310 // Guard against ApplyConfig() being re-entered from its own watchdog
311 // thread (see WatchdogLoop()) -- joining the calling thread would
312 // throw std::system_error ("Resource deadlock avoided"), which
313 // propagates out of the thread function and crashes fppd via
314 // std::terminate(). The watchdog's rebuild path already stops the
315 // watchdog loop and hands off to a short-lived helper thread before
316 // calling ApplyConfig(), but keep this check as a safety net for any
317 // other re-entrant caller.
318
4/4
✓ Branch 0 taken 58 times.
✓ Branch 1 taken 14 times.
✓ Branch 2 taken 58 times.
✓ Branch 3 taken 14 times.
 130 if (m_watchdogThread.joinable() &&
319
1/2
✓ Branch 0 taken 58 times.
✗ Branch 1 not taken.
 58 m_watchdogThread.get_id() != std::this_thread::get_id()) {
320
1/2
✓ Branch 0 taken 58 times.
✗ Branch 1 not taken.
 58 m_watchdogThread.join();
321 }
322
1/2
✓ Branch 0 taken 72 times.
✗ Branch 1 not taken.
 72 StopAllPipelines();
323
3/6
✓ Branch 0 taken 72 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 72 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 72 times.
✗ Branch 5 not taken.
 216 WarningHolder::RemoveWarning(44, OPUS_WARNING_CONFIG_UNREADABLE);
324
325
3/4
✓ Branch 0 taken 72 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 71 times.
 72 if (!FileExists(m_configPath)) {
326
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 LogInfo(VB_MEDIAOUT, "OpusRTPManager: No config file, nothing to apply\n");
327 // No stream is meant to run, so none is failing to start.
328
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ClearOpusPipelineWarnings(true, true);
329 1 m_active.store(false);
330 1 return true;
331 }
332
333
3/4
✓ Branch 0 taken 71 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 69 times.
 71 if (!LoadConfig()) {
334 // The old pipelines are stopped and none replaces them, so nothing
335 // is failing to start: it is the config that is wrong, and that is
336 // what the banner says instead.
337
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 ClearOpusPipelineWarnings(true, true);
338
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 6 WarningHolder::AddWarning(44, OPUS_WARNING_CONFIG_UNREADABLE);
339 2 m_active.store(false);
340 2 return false;
341 }
342
343 69 int enabledCount = 0;
344
2/2
✓ Branch 0 taken 96 times.
✓ Branch 1 taken 69 times.
 165 for (const auto& inst : m_config.instances) {
345
2/2
✓ Branch 0 taken 94 times.
✓ Branch 1 taken 2 times.
 96 if (inst.enabled) enabledCount++;
346 }
347
348
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 65 times.
 69 if (enabledCount == 0) {
349
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 LogInfo(VB_MEDIAOUT, "OpusRTPManager: No enabled instances\n");
350
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 ClearOpusPipelineWarnings(true, true);
351 4 m_active.store(false);
352 4 return true;
353 }
354
355 // Create pipelines for each enabled instance
356 //
357 // Senders held idle because nothing feeds them -- published to
358 // m_deferredSenders below, once the whole pass has run.
359 65 std::map<int, std::string> deferred;
360 65 bool sendFailed = false;
361 65 bool recvFailed = false;
362 65 int failedStarts = 0;
363
364
2/2
✓ Branch 0 taken 94 times.
✓ Branch 1 taken 65 times.
 159 for (const auto& inst : m_config.instances) {
365
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 94 times.
 94 if (!inst.enabled) continue;
366
367
5/8
✓ Branch 0 taken 94 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✓ Branch 3 taken 53 times.
✓ Branch 4 taken 41 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 41 times.
 94 bool wantSend = (inst.mode == "send" || inst.mode == "both");
368
5/8
✓ Branch 0 taken 94 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 53 times.
✓ Branch 3 taken 41 times.
✓ Branch 4 taken 53 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 53 times.
 94 bool wantRecv = (inst.mode == "receive" || inst.mode == "both");
369
370
2/2
✓ Branch 0 taken 53 times.
✓ Branch 1 taken 41 times.
 94 if (wantSend) {
371 // Nothing feeds this sender, so starting it would cost 30 seconds
372 // of blocked apply and end in FAILURE. Hold it instead -- see
373 // OpusRTPConfig::requireGroupSource.
374
2/4
✓ Branch 0 taken 53 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 53 times.
✗ Branch 3 not taken.
 53 const std::string nodeName = SafeNodeName(inst.name) + "_send";
375
7/8
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 9 times.
✓ Branch 2 taken 44 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✓ Branch 5 taken 43 times.
✓ Branch 6 taken 1 times.
✓ Branch 7 taken 52 times.
 53 if (m_config.requireGroupSource && !PipeWireGraphFeedsNode(nodeName)) {
376
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 LogInfo(VB_MEDIAOUT,
377 "Opus RTP send [%d] '%s': nothing in the audio graph feeds %s, "
378 "holding the stream idle. Add it to an Audio Output Group "
379 "and apply that config to start it.\n",
380 inst.id, inst.name.c_str(), nodeName.c_str());
381
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 deferred[inst.id] =
382
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 "Waiting for audio — not a member of any enabled Audio Output Group";
383
3/4
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 49 times.
✓ Branch 3 taken 3 times.
 52 } else if (!CreateSendPipeline(inst)) {
384 49 sendFailed = true;
385 49 failedStarts++;
386 }
387 53 }
388
389
7/8
✓ Branch 0 taken 41 times.
✓ Branch 1 taken 53 times.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 8 times.
✓ Branch 5 taken 33 times.
✓ Branch 6 taken 8 times.
✓ Branch 7 taken 86 times.
 94 if (wantRecv && !CreateRecvPipeline(inst)) {
390 8 recvFailed = true;
391 8 failedStarts++;
392 }
393 }
394
1/2
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
 65 ClearOpusPipelineWarnings(!sendFailed, !recvFailed);
395
396 {
397
1/2
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
 65 std::lock_guard<std::mutex> lock(m_pipelineMutex);
398 65 m_deferredSenders = std::move(deferred);
399 65 m_failedStarts = failedStarts;
400 65 }
401
402 // Start watchdog thread
403 65 m_watchdogRunning.store(true);
404
1/2
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
 65 m_watchdogThread = std::thread(&OpusRTPManager::WatchdogLoop, this);
405
406 65 m_active.store(true);
407
1/2
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
 65 LogInfo(VB_MEDIAOUT, "OpusRTPManager: Applied config -- %d send, %d receive pipelines\n",
408 (int)m_sendPipelines.size(), (int)m_recvPipelines.size());
409 65 return true;
410 65 }
411
412 1 void OpusRTPManager::ScheduleRebuild() {
413
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 std::lock_guard<std::mutex> rl(m_rebuildMutex);
414
3/6
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 1 times.
 1 if (m_stopping.load() || m_rebuildRunning.load()) {
415 // Shutting down, or a rebuild is already on its way. Never join a
416 // running one here: its ApplyConfig() may be waiting to join the
417 // watchdog thread that is calling this.
418 ✗ return;
419 }
420
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (m_rebuildThread.joinable()) {
421 ✗ m_rebuildThread.join(); // finished: returns at once
422 }
423 1 uint64_t generation = m_applyGeneration.load();
424 1 m_rebuildRunning.store(true);
425 1 m_rebuildThread = std::thread([this, generation]() {
426 // Brief delay to let the watchdog thread finish exiting and become
427 // joinable.
428
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 std::this_thread::sleep_for(std::chrono::milliseconds(500));
429 {
430
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 std::lock_guard<std::mutex> applyLock(m_applyMutex);
431 // Nothing to do if anything applied, cleaned up or shut down in
432 // the meantime -- it either did this rebuild's work or undid it
433 // on purpose.
434
3/6
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✗ Branch 5 not taken.
 2 if (!m_stopping.load() && generation == m_applyGeneration.load()) {
435
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ApplyConfigLocked();
436 }
437 1 }
438 1 m_rebuildRunning.store(false);
439
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 2 });
440
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 }
441
442 1 void OpusRTPManager::Cleanup() {
443
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 LogInfo(VB_MEDIAOUT, "OpusRTPManager: Cleanup\n");
444
445
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 std::lock_guard<std::mutex> applyLock(m_applyMutex);
446 // Nothing runs after this, so no stream is failing to start.
447
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ClearOpusPipelineWarnings(true, true);
448
3/6
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✗ Branch 5 not taken.
 3 WarningHolder::RemoveWarning(44, OPUS_WARNING_CONFIG_UNREADABLE);
449 // A watchdog rebuild already scheduled must not bring back what this
450 // tears down.
451 1 m_applyGeneration++;
452
453 1 m_watchdogRunning.store(false);
454
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (m_watchdogThread.joinable()) {
455
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 m_watchdogThread.join();
456 }
457
458
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 StopAllPipelines();
459 1 m_active.store(false);
460 1 }
461
462 // ──────────────────────────────────────────────────────────────────────────────
463 // Helpers
464 // ──────────────────────────────────────────────────────────────────────────────
465 93 bool OpusRTPManager::IsMulticast(const std::string& ip) {
466 struct in_addr addr;
467
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 93 times.
 93 if (inet_pton(AF_INET, ip.c_str(), &addr) != 1) {
468 ✗ return false;
469 }
470 // Multicast range: 224.0.0.0 - 239.255.255.255
471 93 uint32_t hostAddr = ntohl(addr.s_addr);
472
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 93 return (hostAddr >= 0xE0000000 && hostAddr <= 0xEFFFFFFF);
473 }
474
475 187 std::string OpusRTPManager::SafeNodeName(const std::string& name) {
476
1/2
✓ Branch 0 taken 187 times.
✗ Branch 1 not taken.
 187 std::string result = "opusrtp_";
477
2/2
✓ Branch 0 taken 378 times.
✓ Branch 1 taken 187 times.
 565 for (char c : name) {
478
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 378 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 378 if (std::isalnum(c) || c == '_') {
479
1/2
✓ Branch 0 taken 378 times.
✗ Branch 1 not taken.
 378 result += std::tolower(c);
480 } else {
481 ✗ result += '_';
482 }
483 }
484 187 return result;
485 ✗ }
486
487 ✗ std::string OpusRTPManager::GetInterfaceIP(const std::string& iface) {
488 ✗ struct ifaddrs* addrs = nullptr;
489 ✗ std::string result = "0.0.0.0";
490
491 ✗ if (getifaddrs(&addrs) != 0) {
492 ✗ return result;
493 }
494
495 ✗ for (struct ifaddrs* ifa = addrs; ifa; ifa = ifa->ifa_next) {
496 ✗ if (!ifa->ifa_addr || ifa->ifa_addr->sa_family != AF_INET) continue;
497 ✗ if (iface.empty() || iface == ifa->ifa_name) {
498 ✗ if (std::string(ifa->ifa_name) == "lo") continue;
499 ✗ struct sockaddr_in* addr = (struct sockaddr_in*)ifa->ifa_addr;
500 char buf[INET_ADDRSTRLEN];
501 ✗ inet_ntop(AF_INET, &addr->sin_addr, buf, sizeof(buf));
502 ✗ result = buf;
503 ✗ if (!iface.empty()) break;
504 }
505 }
506
507 ✗ freeifaddrs(addrs);
508 ✗ return result;
509 ✗ }
510
511 // ──────────────────────────────────────────────────────────────────────────────
512 // Pipeline creation -- Send
513 // ──────────────────────────────────────────────────────────────────────────────
514 52 bool OpusRTPManager::CreateSendPipeline(const OpusRTPInstance& inst) {
515
2/4
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
 52 std::string nodeName = SafeNodeName(inst.name) + "_send";
516
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 bool multicast = IsMulticast(inst.destIP);
517
518 // Pipeline:
519 // pipewiresrc stream-properties="props,node.name=<node>,node.autoconnect=false"
520 // ! queue ! audioconvert ! audioresample ! audioconvert ! audiorate
521 // ! audio/x-raw,rate=48000,channels=N
522 // ! opusenc bitrate=<bps> dtx=<bool> inband-fec=<bool> packet-loss-percentage=<int>
523 // ! rtpopuspay pt=96
524 // ! udpsink host=<ip> port=<port> qos-dscp=34 [multicast options]
525 //
526 // The resampler is not optional, and audiorate is not a substitute for it:
527 // audiorate only inserts/drops samples to patch timestamp gaps, it cannot
528 // change the rate ("audiorate0 can't handle caps rate=48000"). Opus needs
529 // 48 kHz, and the PipeWire graph does not necessarily run there -- it
530 // follows the output card and sits at 44100 by default -- so without a
531 // resampler this chain has nothing in it that can bridge the two. See the
532 // matching note in AES67Manager::CreateSendPipeline(): the failure mode
533 // when the rates do not line up is that negotiation never completes and
534 // set_state() blocks, not a clean error. The split around audioresample
535 // mirrors what was measured to work there; it costs nothing when the graph
536 // already runs at 48000, where the resampler passes through.
537
538
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 std::ostringstream oss;
539 oss << "pipewiresrc name=pwsrc min-buffers=2 do-timestamp=true "
540 << "! queue max-size-buffers=0 max-size-bytes=0 max-size-time=200000000 leaky=downstream "
541 << "! audioconvert "
542 << "! audioresample "
543 << "! audioconvert "
544 // The master volume (see SetMasterVolume()).
545
9/18
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 52 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 52 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 52 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 52 times.
✗ Branch 13 not taken.
✓ Branch 14 taken 52 times.
✗ Branch 15 not taken.
✓ Branch 16 taken 52 times.
✗ Branch 17 not taken.
 52 << "! volume name=mastervol volume=" << masterVolumeGain(getVolume()) << " "
546 << "! audiorate "
547
4/8
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 52 times.
✗ Branch 7 not taken.
 52 << "! audio/x-raw,rate=" << OpusRTP::AUDIO_RATE
548
2/4
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
 52 << ",channels=" << inst.channels << " "
549
3/6
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
 52 << "! opusenc bitrate=" << inst.bitrate
550 << " frame-size=20"
551 52 << " dtx=" << (inst.dtx ? "true" : "false")
552 52 << " inband-fec=" << (inst.fec ? "true" : "false")
553
9/18
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 52 times.
✓ Branch 6 taken 52 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 52 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 52 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 52 times.
✗ Branch 13 not taken.
✓ Branch 14 taken 52 times.
✗ Branch 15 not taken.
✓ Branch 16 taken 52 times.
✗ Branch 17 not taken.
 52 << " packet-loss-percentage=" << inst.packetLoss << " "
554
3/6
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
 52 << "! rtpopuspay pt=" << OpusRTP::RTP_PAYLOAD_TYPE << " "
555 52 << "! udpsink name=usink host=" << inst.destIP
556
5/10
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 52 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 52 times.
✗ Branch 9 not taken.
 52 << " port=" << inst.port
557
2/4
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
 52 << " qos-dscp=" << OpusRTP::AUDIO_DSCP
558
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 << " sync=false";
559
560
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 52 times.
 52 if (multicast) {
561 ✗ oss << " ttl-mc=" << OpusRTP::MULTICAST_TTL
562 ✗ << " auto-multicast=true";
563 ✗ if (!inst.interface.empty()) {
564 ✗ oss << " multicast-iface=" << inst.interface;
565 }
566 }
567
568
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 std::string pipelineStr = oss.str();
569
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 LogInfo(VB_MEDIAOUT, "Opus RTP send pipeline [%d] %s: %s\n",
570 inst.id, inst.name.c_str(), pipelineStr.c_str());
571
572 52 GError* error = nullptr;
573
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 GstElement* pipeline = gst_parse_launch(pipelineStr.c_str(), &error);
574
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 52 times.
 52 if (error) {
575 ✗ LogErr(VB_MEDIAOUT, "Opus RTP send pipeline error [%d]: %s\n",
576 inst.id, error->message);
577 ✗ g_error_free(error);
578 // gst_parse_launch() can return a non-null (partial) pipeline even
579 // when it also reports an error -- release it so we don't leak.
580 ✗ if (pipeline) {
581 ✗ gst_object_unref(pipeline);
582 }
583 ✗ return false;
584 }
585
586 // Set stream-properties post-launch — inline GstStructure values
587 // containing '/' (e.g. node.latency=960/48000) crash gst_value_deserialize.
588
3/6
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
 52 GstElement* pwsrc = gst_bin_get_by_name(GST_BIN(pipeline), "pwsrc");
589
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 if (pwsrc) {
590
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 GstStructure* props = gst_structure_new("props",
591 "node.name", G_TYPE_STRING, nodeName.c_str(),
592 "node.autoconnect", G_TYPE_BOOLEAN, FALSE,
593 "node.latency", G_TYPE_STRING, "960/48000",
594 NULL);
595
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 g_object_set(pwsrc, "stream-properties", props, NULL);
596
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 gst_structure_free(props);
597
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 gst_object_unref(pwsrc);
598 }
599
600 // On the system clock, not pipewiresrc's GstPipeWireClock, which stops
601 // advancing for a follower stream once the graph's real driver takes it
602 // over; pipewiresrc waits on it before every buffer, and the sender went
603 // quiet after its first packets. Same as RTSPOutputManager's audio bridge.
604
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 GstClock* sysClock = gst_system_clock_obtain();
605
3/6
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
 52 gst_pipeline_use_clock(GST_PIPELINE(pipeline), sysClock);
606
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 gst_object_unref(sysClock);
607
608
3/6
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
✗ Branch 5 not taken.
 52 GstBus* bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
609
610
1/2
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
 52 GstStateChangeReturn ret = gst_element_set_state(pipeline, GST_STATE_PLAYING);
611
2/2
✓ Branch 0 taken 49 times.
✓ Branch 1 taken 3 times.
 52 if (ret == GST_STATE_CHANGE_FAILURE) {
612
1/2
✓ Branch 0 taken 49 times.
✗ Branch 1 not taken.
 49 LogErr(VB_MEDIAOUT, "Opus RTP send pipeline [%d] failed to start\n", inst.id);
613
2/4
✓ Branch 0 taken 49 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 49 times.
✗ Branch 3 not taken.
 147 WarningHolder::AddWarning(44, OPUS_WARNING_SEND_FAILED);
614 // Drop back to NULL before unreffing -- elements may already hold
615 // READY/PAUSED resources (sockets, threads, PipeWire connections)
616 // that gst_object_unref() alone will not release.
617
1/2
✓ Branch 0 taken 49 times.
✗ Branch 1 not taken.
 49 gst_element_set_state(pipeline, GST_STATE_NULL);
618
1/2
✓ Branch 0 taken 49 times.
✗ Branch 1 not taken.
 49 gst_object_unref(bus);
619
1/2
✓ Branch 0 taken 49 times.
✗ Branch 1 not taken.
 49 gst_object_unref(pipeline);
620 49 return false;
621 }
622
623 {
624
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 std::unique_lock<std::mutex> lock(m_pipelineMutex);
625 // If a pipeline already exists for this instance id (e.g. two
626 // ApplyConfig() runs interleaved), stop it properly instead of
627 // erasing the map entry directly -- a raw erase() would destroy
628 // the OpusRTPPipeline (dropping its GstElement* refs) without ever
629 // calling gst_element_set_state(NULL), leaking a still-PLAYING
630 // pipeline (refs, threads, sockets).
631
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 auto node = m_sendPipelines.extract(inst.id);
632
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 if (!node.empty()) {
633 // StopPipeline() can block on a GStreamer state change to
634 // PipeWire -- release the lock while it runs, matching the
635 // no-block-while-locked policy used elsewhere in this file
636 // (see StopAllPipelines()).
637 ✗ lock.unlock();
638 ✗ StopPipeline(node.mapped());
639 ✗ lock.lock();
640 }
641
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 auto [it, ok] = m_sendPipelines.try_emplace(inst.id);
642 3 it->second.instanceId = inst.id;
643 3 it->second.isSend = true;
644 3 it->second.pipeline = pipeline;
645 3 it->second.bus = bus;
646 3 it->second.running = true;
647 // The gain in the description is the one when it was built; a change
648 // since then went to the pipelines already in this map, not this one.
649
2/4
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
 3 SetPipelineMasterVolume(pipeline, getVolume());
650 3 }
651
652
2/4
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
 3 LogInfo(VB_MEDIAOUT, "Opus RTP send pipeline [%d] %s started -> %s:%d (%dch, %d kbps%s)\n",
653 inst.id, inst.name.c_str(), inst.destIP.c_str(), inst.port,
654 inst.channels, inst.bitrate / 1000,
655 multicast ? ", multicast" : ", unicast");
656 3 return true;
657 52 }
658
659 // ──────────────────────────────────────────────────────────────────────────────
660 // Pipeline creation -- Receive
661 // ──────────────────────────────────────────────────────────────────────────────
662 41 bool OpusRTPManager::CreateRecvPipeline(const OpusRTPInstance& inst) {
663
2/4
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
 41 std::string nodeName = SafeNodeName(inst.name) + "_recv";
664
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 bool multicast = IsMulticast(inst.destIP);
665
666 // Pipeline:
667 // udpsrc [multicast-group=<ip>] port=<port> [auto-multicast=true] [multicast-iface=<iface>]
668 // ! application/x-rtp,media=audio,encoding-name=OPUS,payload=96
669 // ! rtpjitterbuffer latency=<ms>
670 // ! rtpopusdepay
671 // ! opusdec
672 // ! audioconvert
673 // ! pipewiresink stream-properties="props,media.class=Audio/Source,node.name=<node>"
674
675
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 std::ostringstream oss;
676
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 oss << "udpsrc";
677
678
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 41 times.
 41 if (multicast) {
679 ✗ oss << " multicast-group=" << inst.destIP;
680 }
681
682
2/4
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
 41 oss << " port=" << inst.port;
683
684
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 41 times.
 41 if (multicast) {
685 ✗ oss << " auto-multicast=true";
686 ✗ if (!inst.interface.empty()) {
687 ✗ oss << " multicast-iface=" << inst.interface;
688 }
689 }
690
691 oss << " ! application/x-rtp,media=audio,encoding-name=OPUS"
692
3/6
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 41 times.
✗ Branch 5 not taken.
 41 << ",payload=" << OpusRTP::RTP_PAYLOAD_TYPE << " "
693
3/6
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 41 times.
✗ Branch 5 not taken.
 41 << "! rtpjitterbuffer latency=" << inst.latency << " "
694 << "! rtpopusdepay "
695 << "! opusdec plc=true "
696 << "! audioconvert "
697
6/12
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 41 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 41 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 41 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 41 times.
✗ Branch 11 not taken.
 41 << "! audio/x-raw,rate=" << OpusRTP::AUDIO_RATE << ",format=F32LE "
698
2/4
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
 41 << "! pipewiresink name=pwsink";
699
700
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 std::string pipelineStr = oss.str();
701
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 LogInfo(VB_MEDIAOUT, "Opus RTP recv pipeline [%d] %s: %s\n",
702 inst.id, inst.name.c_str(), pipelineStr.c_str());
703
704 41 GError* error = nullptr;
705
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 GstElement* pipeline = gst_parse_launch(pipelineStr.c_str(), &error);
706
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 41 times.
 41 if (error) {
707 ✗ LogErr(VB_MEDIAOUT, "Opus RTP recv pipeline error [%d]: %s\n",
708 inst.id, error->message);
709 ✗ g_error_free(error);
710 // gst_parse_launch() can return a non-null (partial) pipeline even
711 // when it also reports an error -- release it so we don't leak.
712 ✗ if (pipeline) {
713 ✗ gst_object_unref(pipeline);
714 }
715 ✗ return false;
716 }
717
718 // Set stream-properties post-launch — gst_parse_launch cannot
719 // deserialize GstStructure values containing '/' (e.g. Audio/Source)
720 // which crashes gst_value_deserialize.
721
2/4
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
 41 std::string recvNodeDesc = SafeNodeName(inst.name) + "_recv";
722
3/6
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 41 times.
✗ Branch 5 not taken.
 41 GstElement* pwsink = gst_bin_get_by_name(GST_BIN(pipeline), "pwsink");
723
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 if (pwsink) {
724
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 GstStructure* props = gst_structure_new("props",
725 "media.class", G_TYPE_STRING, "Audio/Source",
726 "node.name", G_TYPE_STRING, nodeName.c_str(),
727 "node.description", G_TYPE_STRING, recvNodeDesc.c_str(),
728 NULL);
729
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 g_object_set(pwsink, "stream-properties", props, NULL);
730
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 gst_structure_free(props);
731
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 gst_object_unref(pwsink);
732 }
733
734
3/6
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 41 times.
✗ Branch 5 not taken.
 41 GstBus* bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
735
736
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 GstStateChangeReturn ret = gst_element_set_state(pipeline, GST_STATE_PLAYING);
737
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 33 times.
 41 if (ret == GST_STATE_CHANGE_FAILURE) {
738
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 LogErr(VB_MEDIAOUT, "Opus RTP recv pipeline [%d] failed to start\n", inst.id);
739
2/4
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
✗ Branch 3 not taken.
 24 WarningHolder::AddWarning(44, OPUS_WARNING_RECV_FAILED);
740 // Drop back to NULL before unreffing -- elements may already hold
741 // READY/PAUSED resources (sockets, threads, PipeWire connections)
742 // that gst_object_unref() alone will not release.
743
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 gst_element_set_state(pipeline, GST_STATE_NULL);
744
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 gst_object_unref(bus);
745
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 gst_object_unref(pipeline);
746 8 return false;
747 }
748
749 {
750
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 std::unique_lock<std::mutex> lock(m_pipelineMutex);
751 // See CreateSendPipeline() for why we stop any existing pipeline
752 // for this id instead of erasing the map entry directly.
753
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 auto node = m_recvPipelines.extract(inst.id);
754
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 33 times.
 33 if (!node.empty()) {
755 ✗ lock.unlock();
756 ✗ StopPipeline(node.mapped());
757 ✗ lock.lock();
758 }
759
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 auto [it, ok] = m_recvPipelines.try_emplace(inst.id);
760 33 it->second.instanceId = inst.id;
761 33 it->second.isSend = false;
762 33 it->second.pipeline = pipeline;
763 33 it->second.bus = bus;
764 33 it->second.running = true;
765 33 }
766
767
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 LogInfo(VB_MEDIAOUT, "Opus RTP recv pipeline [%d] %s started <- %s:%d (%dch, %dms latency)\n",
768 inst.id, inst.name.c_str(), inst.destIP.c_str(), inst.port,
769 inst.channels, inst.latency);
770 33 return true;
771 41 }
772
773 // ──────────────────────────────────────────────────────────────────────────────
774 // Pipeline management
775 // ──────────────────────────────────────────────────────────────────────────────
776 36 void OpusRTPManager::StopPipeline(OpusRTPPipeline& p) {
777
1/2
✓ Branch 0 taken 36 times.
✗ Branch 1 not taken.
 36 if (p.pipeline) {
778 36 gst_element_set_state(p.pipeline, GST_STATE_NULL);
779
780 // Remove the buffer-drop probe installed by FlushSendPipelines()
781 // (if any) and release our ref on the pad it was installed on.
782 // Without this, the probe stays registered on the pad with
783 // &p.dropCounter as userdata -- a dangling pointer once this
784 // OpusRTPPipeline is erased from its owning map -- and the pad
785 // ref taken in FlushSendPipelines() leaks.
786
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 36 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 36 if (p.probeId != 0 && p.probePad) {
787 ✗ gst_pad_remove_probe(p.probePad, p.probeId);
788 ✗ p.probeId = 0;
789 }
790
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 36 times.
 36 if (p.probePad) {
791 ✗ gst_object_unref(p.probePad);
792 ✗ p.probePad = nullptr;
793 }
794
795
1/2
✓ Branch 0 taken 36 times.
✗ Branch 1 not taken.
 36 if (p.bus) {
796 36 gst_object_unref(p.bus);
797 36 p.bus = nullptr;
798 }
799 36 gst_object_unref(p.pipeline);
800 36 p.pipeline = nullptr;
801 36 p.running = false;
802 }
803 36 }
804
805 86 void OpusRTPManager::StopAllPipelines() {
806 86 std::map<int, OpusRTPPipeline> sendCopy, recvCopy;
807 {
808
1/2
✓ Branch 0 taken 86 times.
✗ Branch 1 not taken.
 86 std::lock_guard<std::mutex> lock(m_pipelineMutex);
809 86 sendCopy.swap(m_sendPipelines);
810 86 recvCopy.swap(m_recvPipelines);
811 // Nothing is configured to be running any more, so no sender is
812 // waiting for a source either. ApplyConfig() refills this after its
813 // create pass; every other caller is a teardown.
814 86 m_deferredSenders.clear();
815 86 }
816
817
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 86 times.
 89 for (auto& [id, p] : sendCopy) {
818
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 LogDebug(VB_MEDIAOUT, "OpusRTPManager: Stopping send pipeline [%d]\n", id);
819
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 StopPipeline(p);
820 }
821
822
2/2
✓ Branch 0 taken 33 times.
✓ Branch 1 taken 86 times.
 119 for (auto& [id, p] : recvCopy) {
823
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 LogDebug(VB_MEDIAOUT, "OpusRTPManager: Stopping recv pipeline [%d]\n", id);
824
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 StopPipeline(p);
825 }
826 86 }
827
828 // Pad probe callback: drops buffers while dropCounter > 0
829 ✗ static GstPadProbeReturn OpusDropBufferProbe(GstPad* pad, GstPadProbeInfo* info, gpointer userData) {
830 ✗ std::atomic<int>* counter = static_cast<std::atomic<int>*>(userData);
831 ✗ if (!(info->type & GST_PAD_PROBE_TYPE_BUFFER))
832 ✗ return GST_PAD_PROBE_OK;
833 ✗ if (*counter <= 0)
834 ✗ return GST_PAD_PROBE_OK;
835 ✗ (*counter)--;
836 ✗ return GST_PAD_PROBE_DROP;
837 }
838
839 3106 void OpusRTPManager::SetMasterVolume(int volume) {
840
1/2
✓ Branch 0 taken 3106 times.
✗ Branch 1 not taken.
 3106 std::lock_guard<std::mutex> lock(m_pipelineMutex);
841
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3106 times.
 3106 for (auto& kv : m_sendPipelines) {
842 ✗ if (kv.second.pipeline) {
843 ✗ SetPipelineMasterVolume(kv.second.pipeline, volume);
844 }
845 }
846 3106 }
847
848 ✗ void OpusRTPManager::FlushSendPipelines() {
849 ✗ std::lock_guard<std::mutex> lock(m_pipelineMutex);
850 ✗ for (auto& kv : m_sendPipelines) {
851 ✗ OpusRTPPipeline& p = kv.second;
852 ✗ if (!p.pipeline || !p.running)
853 ✗ continue;
854
855 ✗ constexpr int DROP_COUNT = 10;
856 ✗ LogInfo(VB_MEDIAOUT, "Opus RTP send pipeline [%d]: dropping next %d buffers\n",
857 p.instanceId, DROP_COUNT);
858
859 ✗ p.dropCounter = DROP_COUNT;
860
861 ✗ if (p.probeId != 0)
862 ✗ continue;
863
864 ✗ GstIterator* it = gst_bin_iterate_sources(GST_BIN(p.pipeline));
865 ✗ GValue val = G_VALUE_INIT;
866 ✗ GstElement* srcElem = nullptr;
867 ✗ if (gst_iterator_next(it, &val) == GST_ITERATOR_OK) {
868 ✗ srcElem = GST_ELEMENT(g_value_get_object(&val));
869 ✗ gst_object_ref(srcElem);
870 ✗ g_value_unset(&val);
871 }
872 ✗ gst_iterator_free(it);
873
874 ✗ if (srcElem) {
875 ✗ GstPad* srcpad = gst_element_get_static_pad(srcElem, "src");
876 ✗ if (srcpad) {
877 ✗ p.probePad = srcpad;
878 ✗ p.probeId = gst_pad_add_probe(
879 srcpad,
880 GST_PAD_PROBE_TYPE_BUFFER,
881 OpusDropBufferProbe,
882 ✗ &p.dropCounter,
883 nullptr);
884 }
885 ✗ gst_object_unref(srcElem);
886 }
887 }
888 ✗ }
889
890 // ──────────────────────────────────────────────────────────────────────────────
891 // Watchdog
892 // ──────────────────────────────────────────────────────────────────────────────
893 65 void OpusRTPManager::WatchdogLoop() {
894 65 constexpr int WATCHDOG_INTERVAL_S = 30;
895
896
1/2
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
 65 while (m_watchdogRunning.load()) {
897 // Sleep in small increments so we can exit promptly
898
6/6
✓ Branch 0 taken 713 times.
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 649 times.
✓ Branch 3 taken 64 times.
✓ Branch 4 taken 649 times.
✓ Branch 5 taken 65 times.
 714 for (int i = 0; i < WATCHDOG_INTERVAL_S * 10 && m_watchdogRunning.load(); i++) {
899
1/2
✓ Branch 0 taken 649 times.
✗ Branch 1 not taken.
 649 std::this_thread::sleep_for(std::chrono::milliseconds(100));
900 }
901
902
2/2
✓ Branch 0 taken 64 times.
✓ Branch 1 taken 1 times.
 65 if (!m_watchdogRunning.load()) break;
903
904
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (PollPipelinesWatchdog()) {
905 // We cannot call ApplyConfig() directly from this thread:
906 // ApplyConfig() joins m_watchdogThread -- which IS this thread
907 // -- and std::thread::join() on the calling thread itself
908 // throws std::system_error, which propagates out of this
909 // thread function and terminates fppd (see AES67Manager's
910 // SAPAnnounceLoop() for the same issue). Instead, stop this
911 // loop and hand off to a short-lived thread that calls
912 // ApplyConfig() once we've exited.
913 1 LogWarn(VB_MEDIAOUT, "OpusRTPManager: Watchdog detected failed pipelines, scheduling rebuild\n");
914 1 m_watchdogRunning.store(false);
915 // Track this as m_rebuildThread (rather than detaching) so
916 // Shutdown() can join it before tearing anything else down -- a
917 // detached thread could otherwise call ApplyConfig() and
918 // resurrect pipelines after Shutdown() has already run, and a
919 // detached thread racing Shutdown()'s own join of
920 // m_watchdogThread is undefined behavior.
921 1 ScheduleRebuild();
922 1 break;
923 }
924 }
925 65 }
926
927 1 bool OpusRTPManager::PollPipelinesWatchdog() {
928 1 bool needsRebuild = false;
929
930
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 std::lock_guard<std::mutex> lock(m_pipelineMutex);
931
932 2 auto checkPipelines = [this, &needsRebuild](std::map<int, OpusRTPPipeline>& pipelines, const char* direction) {
933
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 for (auto& kv : pipelines) {
934 ✗ OpusRTPPipeline& p = kv.second;
935 ✗ if (!p.pipeline || !p.running) continue;
936
937 // Drain bus messages
938 ✗ if (p.bus) {
939 GstMessage* msg;
940 ✗ while ((msg = gst_bus_pop(p.bus)) != nullptr) {
941 ✗ switch (GST_MESSAGE_TYPE(msg)) {
942 ✗ case GST_MESSAGE_ERROR: {
943 ✗ GError* err = nullptr; gchar* dbg = nullptr;
944 ✗ gst_message_parse_error(msg, &err, &dbg);
945 ✗ LogErr(VB_MEDIAOUT,
946 "Opus RTP %s pipeline [%d] bus error: %s\n",
947 direction, p.instanceId, err->message);
948 ✗ g_error_free(err); g_free(dbg);
949 ✗ p.errorMessage = "GStreamer error";
950 ✗ break;
951 }
952 ✗ case GST_MESSAGE_WARNING: {
953 ✗ GError* err = nullptr; gchar* dbg = nullptr;
954 ✗ gst_message_parse_warning(msg, &err, &dbg);
955 ✗ LogWarn(VB_MEDIAOUT,
956 "Opus RTP %s pipeline [%d] bus warning: %s\n",
957 direction, p.instanceId, err->message);
958 ✗ g_error_free(err); g_free(dbg);
959 ✗ break;
960 }
961 ✗ default: break;
962 }
963 ✗ gst_message_unref(msg);
964 }
965 }
966
967 // Check current pipeline state
968 ✗ GstState curState = GST_STATE_NULL, pendingState = GST_STATE_VOID_PENDING;
969 ✗ gst_element_get_state(p.pipeline, &curState, &pendingState, 0);
970
971 ✗ if (curState != GST_STATE_PLAYING && pendingState != GST_STATE_PLAYING) {
972 ✗ LogWarn(VB_MEDIAOUT,
973 "Opus RTP %s pipeline [%d] is in %s state -- flagging for rebuild\n",
974 direction, p.instanceId,
975 gst_element_state_get_name(curState));
976 ✗ p.running = false;
977 ✗ p.errorMessage = "Watchdog: not in PLAYING state";
978 ✗ needsRebuild = true;
979 ✗ } else if (p.isSend) {
980 // Zombie detection for send pipelines: check udpsink byte count
981 ✗ GstElement* usink = gst_bin_get_by_name(GST_BIN(p.pipeline), "usink");
982 ✗ if (usink) {
983 ✗ guint64 bytesSent = 0;
984 ✗ g_object_get(usink, "bytes-served", &bytesSent, nullptr);
985 ✗ gst_object_unref(usink);
986
987 ✗ if (bytesSent == p.lastByteCount) {
988 ✗ p.stallCount++;
989 ✗ if (p.stallCount >= 2) {
990 ✗ LogWarn(VB_MEDIAOUT,
991 "Opus RTP send pipeline [%d] stalled (no bytes for %d checks)\n",
992 p.instanceId, p.stallCount);
993 ✗ p.running = false;
994 ✗ p.errorMessage = "Watchdog: send pipeline stalled";
995 ✗ needsRebuild = true;
996 }
997 } else {
998 ✗ p.stallCount = 0;
999 }
1000 ✗ p.lastByteCount = bytesSent;
1001 }
1002 }
1003 }
1004 2 };
1005
1006
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 checkPipelines(m_sendPipelines, "send");
1007
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 checkPipelines(m_recvPipelines, "recv");
1008
1009 // A stream that failed to start is never stored, so nothing above sees
1010 // it. The first retry is this pass; one that keeps failing waits 30s,
1011 // 60s, ... up to 8 minutes, and the count starts over after 10 quiet
1012 // minutes.
1013
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (m_failedStarts > 0) {
1014 1 const auto now = std::chrono::steady_clock::now();
1015
2/10
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
✓ Branch 9 taken 1 times.
 1 if (m_failedStartRebuilds > 0 && now - m_lastFailedStartRebuild > std::chrono::minutes(10)) {
1016 ✗ m_failedStartRebuilds = 0;
1017 }
1018 1 const auto wait = std::chrono::seconds(30) * (1 << std::min(std::max(m_failedStartRebuilds - 1, 0), 4));
1019
3/12
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 1 times.
✗ Branch 11 not taken.
 1 if (needsRebuild || m_failedStartRebuilds == 0 || now - m_lastFailedStartRebuild >= wait) {
1020
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (!needsRebuild) {
1021 1 m_failedStartRebuilds++;
1022 1 m_lastFailedStartRebuild = now;
1023 }
1024 1 needsRebuild = true;
1025
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 LogWarn(VB_MEDIAOUT, "Opus RTP watchdog: %d stream(s) failed to start, trying again\n",
1026 m_failedStarts);
1027 }
1028 }
1029
1030 1 return needsRebuild;
1031 1 }
1032
1033 584 OpusRTPManager::Status OpusRTPManager::GetStatus() {
1034 584 Status status;
1035
1036 // Snapshot the config BEFORE taking the pipeline lock and use the copy
1037 // from here on -- reading m_config directly would race LoadConfig()
1038 // reallocating the instance vector on another thread. See m_configMutex.
1039
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 OpusRTPConfig config = GetConfigSnapshot();
1040
1041 584 std::unique_lock<std::mutex> lock(m_pipelineMutex, std::try_to_lock);
1042
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 if (lock.owns_lock()) {
1043
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 584 times.
 586 for (const auto& [id, p] : m_sendPipelines) {
1044 2 Status::PipelineStatus ps;
1045 2 ps.instanceId = id;
1046
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 ps.mode = "send";
1047 2 ps.running = p.running;
1048
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 ps.error = p.errorMessage;
1049
1050
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 for (const auto& inst : config.instances) {
1051
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 if (inst.id == id) {
1052
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 ps.name = inst.name;
1053 2 break;
1054 }
1055 }
1056
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 status.pipelines.push_back(ps);
1057 2 }
1058
1059
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 584 times.
 585 for (const auto& [id, p] : m_recvPipelines) {
1060 1 Status::PipelineStatus ps;
1061 1 ps.instanceId = id;
1062
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ps.mode = "receive";
1063 1 ps.running = p.running;
1064
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ps.error = p.errorMessage;
1065
1066
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 for (const auto& inst : config.instances) {
1067
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 1 times.
 2 if (inst.id == id) {
1068
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ps.name = inst.name;
1069 1 break;
1070 }
1071 }
1072
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 status.pipelines.push_back(ps);
1073 1 }
1074
1075 // Senders deliberately held idle. Reported alongside the real
1076 // pipelines rather than omitted: a stream the user enabled and cannot
1077 // see anywhere reads as FPP having lost the config.
1078
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 584 times.
 585 for (const auto& [id, reason] : m_deferredSenders) {
1079 1 Status::PipelineStatus ps;
1080 1 ps.instanceId = id;
1081
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ps.mode = "send";
1082 1 ps.running = false;
1083 1 ps.waitingForSource = true;
1084
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ps.note = reason;
1085
1086
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 for (const auto& inst : config.instances) {
1087
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (inst.id == id) {
1088
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 ps.name = inst.name;
1089 1 break;
1090 }
1091 }
1092
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 status.pipelines.push_back(ps);
1093 1 }
1094 } else {
1095 ✗ Status::PipelineStatus ps;
1096 ✗ ps.instanceId = -1;
1097 ✗ ps.name = "(status unavailable -- pipeline operation in progress)";
1098 ✗ ps.mode = "unknown";
1099 ✗ ps.running = false;
1100 ✗ status.pipelines.push_back(ps);
1101 ✗ }
1102
1103 584 return status;
1104 584 }
1105
1106 // ──────────────────────────────────────────────────────────────────────────────
1107 // HTTP API
1108 // ──────────────────────────────────────────────────────────────────────────────
1109 584 HttpResponsePtr OpusRTPManager::render_GET(const HttpRequestPtr& req) {
1110
2/4
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 584 times.
✗ Branch 3 not taken.
 584 std::string url(req->path());
1111
1112
2/2
✓ Branch 0 taken 579 times.
✓ Branch 1 taken 5 times.
 584 if (url.find("/opusrtp/") == 0) {
1113
1/2
✓ Branch 0 taken 579 times.
✗ Branch 1 not taken.
 579 url = url.substr(9);
1114
2/4
✓ Branch 0 taken 5 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5 times.
✗ Branch 3 not taken.
 5 } else if (url == "/opusrtp") {
1115
1/2
✓ Branch 0 taken 5 times.
✗ Branch 1 not taken.
 5 url = "status";
1116 }
1117
1118
2/4
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 584 times.
✗ Branch 3 not taken.
 584 if (url == "status") {
1119
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 Status st = GetStatus();
1120
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 Json::Value result;
1121
1122
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 Json::Value pipelines(Json::arrayValue);
1123
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 584 times.
 588 for (const auto& p : st.pipelines) {
1124
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 Json::Value pj;
1125
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 4 pj["instanceId"] = p.instanceId;
1126
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 4 pj["name"] = p.name;
1127
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 4 pj["mode"] = p.mode;
1128
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 4 pj["running"] = p.running;
1129
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 4 times.
 4 if (!p.error.empty()) {
1130 ✗ pj["error"] = p.error;
1131 }
1132 // Distinct from "error" on purpose: the UI renders a failure in
1133 // red and this as guidance, so collapsing the two would put a
1134 // brand new instance back in the alarming state this replaced.
1135
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 3 times.
 4 if (p.waitingForSource) {
1136
2/4
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
 1 pj["waitingForSource"] = true;
1137 }
1138
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 3 times.
 4 if (!p.note.empty()) {
1139
2/4
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
 1 pj["note"] = p.note;
1140 }
1141
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 pipelines.append(pj);
1142 4 }
1143
2/4
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 584 times.
✗ Branch 3 not taken.
 584 result["pipelines"] = pipelines;
1144
2/4
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 584 times.
✗ Branch 3 not taken.
 584 result["active"] = m_active.load();
1145
1146
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 Json::StreamWriterBuilder wbuilder;
1147
3/6
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 584 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 584 times.
✗ Branch 5 not taken.
 1752 wbuilder["indentation"] = "";
1148
1/2
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
 584 std::string resultStr = Json::writeString(wbuilder, result);
1149
1150
2/4
✓ Branch 0 taken 584 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 584 times.
✗ Branch 3 not taken.
 584 return makeStringResponse(resultStr, 200, "application/json");
1151 584 }
1152
1153 ✗ return makeStringResponse("{\"error\":\"unknown endpoint\"}", 404, "application/json");
1154 584 }
1155
1156 #endif // HAS_OPUS_RTP_GSTREAMER
1157