GCC Code Coverage Report

Directory: ./
File: mediaoutput/OpusRTPManager.h
Date: 2026-10-29 08:11:58
Exec Total Coverage
Lines: 2 2 100.0%
Functions: 1 1 100.0%
Branches: 2 4 50.0%
Line Branch Exec Source
1 #pragma once
2 /*
3 * This file is part of the Falcon Player (FPP) and is Copyright (C)
4 * 2013-2025 by the Falcon Player Developers.
5 *
6 * The Falcon Player (FPP) is free software, and is covered under
7 * multiple Open Source licenses. Please see the included 'LICENSES'
8 * file for descriptions of what files are covered by each license.
9 *
10 * This source file is covered under the LGPL v2.1 as described in the
11 * included LICENSE.LGPL file.
12 */
13
14 // OpusRTPManager -- GStreamer-based Opus RTP audio send/receive
15 //
16 // Provides WiFi-friendly compressed audio streaming using the Opus codec
17 // over RTP, as an alternative to AES67 (uncompressed L24) which requires
18 // wired Ethernet. Supports both unicast and multicast destinations.
19 //
20 // Features:
21 // - Send: pipewiresrc -> audioconvert -> opusenc -> rtpopuspay -> udpsink
22 // - Receive: udpsrc -> rtpjitterbuffer -> rtpopusdepay -> opusdec -> audioconvert -> pipewiresink
23 // - Configurable bitrate, DTX, in-band FEC
24 // - Unicast or multicast destination
25 // - Config format: pipewire-opus-rtp-instances.json
26
27 #if __has_include(<gst/gst.h>)
28 #define HAS_OPUS_RTP_GSTREAMER
29
30 #include <gst/gst.h>
31
32 #include "fpphttp_types.h"
33
34 #include <atomic>
35 #include <chrono>
36 #include <cstdint>
37 #include <map>
38 #include <mutex>
39 #include <string>
40 #include <thread>
41 #include <vector>
42
43 // ──────────────────────────────────────────────────────────────────────────────
44 // Opus RTP Protocol Constants
45 // ──────────────────────────────────────────────────────────────────────────────
46 namespace OpusRTP {
47
48 constexpr int RTP_PAYLOAD_TYPE = 96;
49 constexpr int AUDIO_RATE = 48000; // Opus native rate
50 constexpr int DEFAULT_PORT = 5005; // avoid AES67's 5004
51 constexpr int DEFAULT_CHANNELS = 2;
52 constexpr int DEFAULT_LATENCY_MS = 50; // higher default for WiFi
53 constexpr int DEFAULT_BITRATE = 128000; // 128 kbps
54 constexpr int DEFAULT_PACKET_LOSS = 5; // 5% expected loss
55 constexpr int MULTICAST_TTL = 4;
56
57 // DSCP codepoint for the RTP audio stream (udpsink qos-dscp). AF41, the same
58 // marking AES67 audio uses. It matters more here than there: this is the WiFi
59 // transport, and WiFi maps DSCP onto WMM access categories (AF41 -> AC_VI), so
60 // the marking decides how the stream contends on a busy AP.
61 constexpr int AUDIO_DSCP = 34; // AF41
62
63 constexpr const char* DEFAULT_DEST_IP = "239.69.1.1";
64
65 // Valid bitrates (bps)
66 96 inline bool IsValidBitrate(int br) {
67
2/4
✓ Branch 0 taken 96 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 96 times.
✗ Branch 3 not taken.
 96 return br >= 6000 && br <= 512000;
68 }
69
70 } // namespace OpusRTP
71
72 // ──────────────────────────────────────────────────────────────────────────────
73 // Instance configuration -- mirrors the JSON schema
74 // ──────────────────────────────────────────────────────────────────────────────
75 struct OpusRTPInstance {
76 int id = 0;
77 std::string name;
78 bool enabled = true;
79 std::string mode = "send"; // "send" | "receive" | "both"
80 std::string destIP = OpusRTP::DEFAULT_DEST_IP; // unicast or multicast
81 int port = OpusRTP::DEFAULT_PORT;
82 int channels = OpusRTP::DEFAULT_CHANNELS;
83 std::string interface; // network interface (e.g., "wlan0")
84 int latency = OpusRTP::DEFAULT_LATENCY_MS;
85 int bitrate = OpusRTP::DEFAULT_BITRATE;
86 bool dtx = false; // Discontinuous Transmission
87 bool fec = true; // In-band Forward Error Correction
88 int packetLoss = OpusRTP::DEFAULT_PACKET_LOSS; // expected packet loss %
89 };
90
91 struct OpusRTPConfig {
92 std::vector<OpusRTPInstance> instances;
93
94 // Hold a sender idle until something in the audio graph actually feeds it.
95 //
96 // A send instance is a PipeWire *sink*: its pipewiresrc is created with
97 // node.autoconnect=false, so nothing links into it unless an Audio Output
98 // Group member names it as node.target. With nothing linked the pipeline
99 // cannot preroll, and gst_element_set_state() sits there for 30 seconds
100 // before returning FAILURE -- per instance, on every Apply. That is the
101 // chicken and egg a new instance walks straight into: it cannot be added
102 // to a group until it has been saved, and saving it costs half a minute of
103 // dead UI and a "stream failed to start" warning for a stream the user has
104 // not finished configuring yet.
105 //
106 // So when nothing feeds it, do not start it at all: the instance stays
107 // configured, ApplyConfig() returns promptly, GetStatus() reports it as
108 // waiting rather than failed, and the next apply -- which the Audio Output
109 // Groups page performs anyway, by restarting the stack -- starts it for
110 // real. See PipeWireGraphFeedsNode().
111 //
112 // Set false in pipewire-opus-rtp-instances.json for a graph FPP did not
113 // generate (a hand-written PipeWire conf, or nodes linked by hand with
114 // pw-link), where the group config cannot answer the question.
115 bool requireGroupSource = true;
116 };
117
118 // ──────────────────────────────────────────────────────────────────────────────
119 // Pipeline wrapper -- one per send or receive direction of an instance
120 // ──────────────────────────────────────────────────────────────────────────────
121 struct OpusRTPPipeline {
122 int instanceId = 0;
123 bool isSend = true;
124 GstElement* pipeline = nullptr;
125 GstBus* bus = nullptr;
126
127 bool running = false;
128 std::string errorMessage;
129
130 // Buffer-drop probe for flush between tracks
131 std::atomic<int> dropCounter = 0;
132 GstPad* probePad = nullptr;
133 gulong probeId = 0;
134
135 // Zombie detection
136 uint64_t lastByteCount = 0;
137 int stallCount = 0;
138 };
139
140 // ──────────────────────────────────────────────────────────────────────────────
141 // OpusRTPManager -- singleton managing all Opus RTP GStreamer pipelines
142 // ──────────────────────────────────────────────────────────────────────────────
143 class OpusRTPManager {
144 public:
145 static OpusRTPManager& INSTANCE;
146
147 // HTTP API endpoint (registered at /opusrtp)
148 HttpResponsePtr render_GET(const HttpRequestPtr& req);
149
150 // Lifecycle
151 bool Init();
152 void Shutdown();
153
154 // Apply configuration -- reads JSON and rebuilds pipelines
155 bool ApplyConfig();
156
157 // Cleanup -- stop all pipelines
158 void Cleanup();
159
160 // Status query
161 struct Status {
162 struct PipelineStatus {
163 int instanceId;
164 std::string name;
165 std::string mode;
166 bool running;
167 std::string error;
168 // Configured and enabled, but deliberately not started because
169 // nothing in the audio graph feeds it -- see
170 // OpusRTPConfig::requireGroupSource. Not an error: `note` carries
171 // the reason for the UI to render as guidance rather than a
172 // failure.
173 bool waitingForSource = false;
174 std::string note;
175 };
176 std::vector<PipelineStatus> pipelines;
177 };
178 Status GetStatus();
179
180 // Check if manager is active
181 bool IsActive() const { return m_active.load(); }
182
183 // Flush send pipelines (media playback integration)
184 void FlushSendPipelines();
185
186 // Apply the master volume to every send pipeline (see
187 // AES67Manager::SetMasterVolume()).
188 void SetMasterVolume(int volume);
189
190 OpusRTPManager();
191 virtual ~OpusRTPManager();
192
193 private:
194 // Config
195 OpusRTPConfig m_config;
196 std::string m_configPath;
197 bool LoadConfig();
198
199 // Guards every access to m_config.
200 //
201 // LoadConfig() clears and refills m_config.instances, which reallocates
202 // the vector and reassigns its std::strings. ApplyConfig() is safe (it
203 // holds m_applyMutex and joins the watchdog first), but GetStatus() runs
204 // on an HTTP thread and iterates the same vector -- and UIs poll status,
205 // so a reload landing mid-poll is a use-after-free.
206 //
207 // Held only for the duration of a read or the final swap in LoadConfig(),
208 // and NEVER together with m_pipelineMutex -- GetStatus() snapshots the
209 // config, releases, then takes the pipeline lock. The two locks are
210 // therefore never held at once and cannot deadlock against each other.
211 std::mutex m_configMutex;
212 OpusRTPConfig GetConfigSnapshot();
213
214 // Pipeline watchdog -- polls bus messages and recovers crashed pipelines
215 bool PollPipelinesWatchdog();
216 // Streams whose pipeline failed to build or start in the last apply
217 // (PipeWire not up yet, say). Nothing else would try them again, so the
218 // watchdog rebuilds for them, waiting 30s, 60s, ... up to 8 minutes
219 // between tries: a rebuild restarts every stream. m_failedStarts is set
220 // by ApplyConfig() under m_pipelineMutex; the rest is watchdog thread only.
221 int m_failedStarts = 0;
222 int m_failedStartRebuilds = 0;
223 std::chrono::steady_clock::time_point m_lastFailedStartRebuild{};
224
225 // Watchdog thread
226 std::thread m_watchdogThread;
227 std::atomic<bool> m_watchdogRunning{false};
228 void WatchdogLoop();
229
230 // Deferred pipeline-rebuild thread -- spawned by WatchdogLoop()'s rebuild
231 // path to call ApplyConfig() from a thread other than the one
232 // ApplyConfig() needs to join (m_watchdogThread). Tracked as a member
233 // (rather than detached) so Shutdown() can join it before tearing
234 // anything else down -- a detached thread could otherwise call
235 // ApplyConfig() and resurrect pipelines after Shutdown() has already run,
236 // and a detached thread racing Shutdown()'s own join of m_watchdogThread
237 // is undefined behavior.
238 std::thread m_rebuildThread;
239
240 // Rebuild bookkeeping. m_rebuildMutex guards m_rebuildThread. A rebuild
241 // is scheduled with the apply generation it was asked for and runs only
242 // if nothing has applied, cleaned up or shut down since (see
243 // ScheduleRebuild()); m_stopping is set once by Shutdown() and keeps
244 // ApplyConfig() from bringing anything back after it.
245 std::mutex m_rebuildMutex;
246 std::atomic<bool> m_rebuildRunning{false};
247 std::atomic<bool> m_stopping{false};
248 std::atomic<uint64_t> m_applyGeneration{0};
249 void ScheduleRebuild();
250 bool ApplyConfigLocked();
251
252 // Serializes the whole tear-down/rebuild sequence in ApplyConfig(),
253 // Shutdown() and Cleanup(). ApplyConfig() joins the watchdog thread at
254 // the top and re-creates it ~50 lines later, after loading config and
255 // building pipelines; that window is wide, and ApplyConfig() has three
256 // independent callers (the boot sequence, OpusRTPApplyCommand::run on a
257 // command thread, and m_rebuildThread from the watchdog). Two overlapping calls both pass the join, then both assign
258 // m_watchdogThread -- and assigning over a still-joinable std::thread is
259 // an unconditional std::terminate().
260 //
261 // Shutdown() must join m_rebuildThread BEFORE taking this lock: that
262 // thread calls ApplyConfig(), so holding the lock across the join would
263 // deadlock.
264 std::mutex m_applyMutex;
265
266 // Pipeline management
267 std::map<int, OpusRTPPipeline> m_sendPipelines;
268 std::map<int, OpusRTPPipeline> m_recvPipelines;
269 std::mutex m_pipelineMutex;
270
271 // Send instances ApplyConfig() chose not to start because nothing feeds
272 // them -- see OpusRTPConfig::requireGroupSource. Instance ID to the
273 // reason, which is shown to the user, so phrase it as guidance rather than
274 // as an error. Guarded by m_pipelineMutex alongside the pipeline maps, so
275 // a status reader sees the whole picture -- running, failed and waiting --
276 // under one lock.
277 std::map<int, std::string> m_deferredSenders;
278
279 bool CreateSendPipeline(const OpusRTPInstance& inst);
280 bool CreateRecvPipeline(const OpusRTPInstance& inst);
281 void StopPipeline(OpusRTPPipeline& p);
282 void StopAllPipelines();
283
284 // Get source IP for the network interface
285 std::string GetInterfaceIP(const std::string& iface);
286
287 // Check if destination IP is multicast
288 static bool IsMulticast(const std::string& ip);
289
290 // Safe node name
291 static std::string SafeNodeName(const std::string& name);
292
293 // State
294 std::atomic<bool> m_active{false};
295 std::atomic<bool> m_initialized{false};
296 };
297
298 #endif // HAS_OPUS_RTP_GSTREAMER
299