From d43b9f5dda6150c56850ac5ea814e9b6133ac029 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 7 Oct 2026 11:01:51 -0500 Subject: [PATCH 1/6] Share the PipeWire registry walk and the Pulse query helpers Lifted from the sink sources into their headers so the capture backends can list source nodes and query the server the same way. --- src/pipewire_sink.cpp | 47 ++++++++++--------------------- src/pipewire_sink.h | 29 +++++++++++++++++++ src/pulse_sink.cpp | 65 +++++++++++-------------------------------- src/pulse_sink.h | 36 ++++++++++++++++++++++++ 4 files changed, 96 insertions(+), 81 deletions(-) diff --git a/src/pipewire_sink.cpp b/src/pipewire_sink.cpp index 1d33aee..ccb5a06 100644 --- a/src/pipewire_sink.cpp +++ b/src/pipewire_sink.cpp @@ -67,32 +67,10 @@ bool spa_format_for(uint8_t bits_per_sample, spa_audio_format& format) { return false; } -/// Holds the thread-loop lock for a scope; it does not hold off the realtime data thread. -class LoopLock { -public: - explicit LoopLock(pw_thread_loop* loop) : loop_(loop) { - pw_thread_loop_lock(this->loop_); - } - ~LoopLock() { - pw_thread_loop_unlock(this->loop_); - } - - LoopLock(const LoopLock&) = delete; - LoopLock& operator=(const LoopLock&) = delete; - -private: - pw_thread_loop* loop_; -}; - -/// One audio sink node as the graph describes it. -struct PipeWireNode { - std::string name; - std::string description; -}; - /// The registry walk's state, shared between its callbacks and the thread waiting on them. struct RegistryWalk { pw_thread_loop* loop{nullptr}; + std::vector media_classes; std::vector nodes; int sync_seq{0}; bool done{false}; @@ -108,13 +86,17 @@ void registry_global_cb(void* data, uint32_t /*id*/, uint32_t /*permissions*/, c return; } const char* media_class = spa_dict_lookup(props, PW_KEY_MEDIA_CLASS); - // Only Audio/Sink nodes can take a stream. - if (media_class == nullptr || std::strcmp(media_class, "Audio/Sink") != 0) { + const bool wanted = media_class != nullptr && + std::any_of(walk->media_classes.begin(), walk->media_classes.end(), + [media_class](const char* candidate) { + return std::strcmp(media_class, candidate) == 0; + }); + if (!wanted) { return; } const char* node_name = spa_dict_lookup(props, PW_KEY_NODE_NAME); if (node_name == nullptr) { - return; // nothing -o could name + return; // nothing a device spec could name } const char* description = spa_dict_lookup(props, PW_KEY_NODE_DESCRIPTION); if (description == nullptr) { @@ -159,8 +141,10 @@ constexpr pw_core_events CORE_EVENTS = [] { return events; }(); -/// Walks the daemon's registry once on a private loop and reports every audio sink node. -bool walk_registry(std::vector& out, std::string& error) { +} // namespace + +bool pipewire_list_nodes(std::initializer_list media_classes, + std::vector& out, std::string& error) { pw_thread_loop* loop = pw_thread_loop_new("sendspin-pw-scan", nullptr); if (loop == nullptr) { error = "cannot create a PipeWire loop"; @@ -174,6 +158,7 @@ bool walk_registry(std::vector& out, std::string& error) { RegistryWalk walk; walk.loop = loop; + walk.media_classes = media_classes; pw_context* context = nullptr; pw_core* core = nullptr; pw_registry* registry = nullptr; @@ -246,8 +231,6 @@ bool walk_registry(std::vector& out, std::string& error) { return true; } -} // namespace - size_t pipewire_ring_frames(uint32_t rate, uint32_t buffer_ms) { const auto by_time = static_cast((static_cast(rate) * buffer_ms) / 1000); return std::max(by_time, MIN_RING_FRAMES); @@ -305,7 +288,7 @@ bool PipeWireSink::probe(const std::string& device, std::string& error) { const PipeWireGuard guard; std::vector nodes; - if (!walk_registry(nodes, error)) { + if (!pipewire_list_nodes({"Audio/Sink"}, nodes, error)) { error += " -- run with -l to see what this host has, or -o alsa:pipewire to reach the same " "graph through ALSA's plugin PCM"; return false; @@ -329,7 +312,7 @@ void PipeWireSink::list_devices(std::FILE* out) { std::vector nodes; std::string error; - if (!walk_registry(nodes, error)) { + if (!pipewire_list_nodes({"Audio/Sink"}, nodes, error)) { std::fprintf(out, " (%s)\n", error.c_str()); return; } diff --git a/src/pipewire_sink.h b/src/pipewire_sink.h index 935cf55..b76720d 100644 --- a/src/pipewire_sink.h +++ b/src/pipewire_sink.h @@ -28,6 +28,7 @@ #include #include #include +#include #include #include #include @@ -76,6 +77,34 @@ class PipeWireGuard { PipeWireGuard& operator=(const PipeWireGuard&) = delete; }; +/// Holds the thread-loop lock for a scope; it does not hold off the realtime data thread. +class LoopLock { +public: + explicit LoopLock(pw_thread_loop* loop) : loop_(loop) { + pw_thread_loop_lock(this->loop_); + } + ~LoopLock() { + pw_thread_loop_unlock(this->loop_); + } + + LoopLock(const LoopLock&) = delete; + LoopLock& operator=(const LoopLock&) = delete; + +private: + pw_thread_loop* loop_; +}; + +/// One audio node as the graph describes it. +struct PipeWireNode { + std::string name; + std::string description; +}; + +/// Walks the daemon's registry once on a private loop for nodes of any of `media_classes`. +/// Needs a live PipeWireGuard. +bool pipewire_list_nodes(std::initializer_list media_classes, + std::vector& out, std::string& error); + /// An AudioSink that plays through a PipeWire graph; `-o pipewire:` a node name or empty. /// process() reads the ring and format fields unlocked: change them only while disconnected. /// Lock order is mutex_ then the thread-loop lock; no PipeWire callback takes mutex_. diff --git a/src/pulse_sink.cpp b/src/pulse_sink.cpp index d6b8871..8949d21 100644 --- a/src/pulse_sink.cpp +++ b/src/pulse_sink.cpp @@ -64,37 +64,6 @@ bool pulse_format_for(uint8_t bits_per_sample, pa_sample_format_t& format) { return false; } -/// Holds the libpulse mainloop lock for a scope. Every pa_* call on a live context needs it. -class MainloopLock { -public: - explicit MainloopLock(pa_threaded_mainloop* loop) : loop_(loop) { - pa_threaded_mainloop_lock(this->loop_); - } - ~MainloopLock() { - pa_threaded_mainloop_unlock(this->loop_); - } - - MainloopLock(const MainloopLock&) = delete; - MainloopLock& operator=(const MainloopLock&) = delete; - -private: - pa_threaded_mainloop* loop_; -}; - -/// State shared by an asynchronous server query and its waiter; `done` is read unlocked. -struct PulseQuery { - PulseConnection* conn{nullptr}; - std::atomic done{false}; - bool failed{false}; - - /// Called from the query's own callback, on the mainloop thread, exactly once. - void finish(bool ok) { - this->failed = !ok; - this->done.store(true, std::memory_order_release); - this->conn->notify(); - } -}; - /// One sink as the server describes it. struct PulseSinkInfo { std::string name; @@ -135,24 +104,6 @@ void server_info_cb(pa_context* /*context*/, const pa_server_info* info, void* u query->finish(info != nullptr); } -/// Waits for one query, cancelling it on timeout so a late callback cannot touch a dead `query`. -/// @return true if the query completed and the server answered it. -bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op) { - if (op == nullptr) { - return false; - } - const bool answered = - conn.wait_for([&query] { return query.done.load(std::memory_order_acquire); }); - { - const MainloopLock lock(conn.mainloop()); - if (!answered) { - pa_operation_cancel(op); - } - pa_operation_unref(op); - } - return answered && !query.failed; -} - /// Asks the server for every sink it has. The mainloop lock must NOT be held. bool list_sinks(PulseConnection& conn, std::vector& out) { SinkListQuery query; @@ -186,6 +137,22 @@ std::string default_sink_name(PulseConnection& conn) { } // namespace +bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op) { + if (op == nullptr) { + return false; + } + const bool answered = + conn.wait_for([&query] { return query.done.load(std::memory_order_acquire); }); + { + const MainloopLock lock(conn.mainloop()); + if (!answered) { + pa_operation_cancel(op); + } + pa_operation_unref(op); + } + return answered && !query.failed; +} + PulseConnection::~PulseConnection() { this->disconnect(); } diff --git a/src/pulse_sink.h b/src/pulse_sink.h index 04f86e9..a9cc118 100644 --- a/src/pulse_sink.h +++ b/src/pulse_sink.h @@ -96,6 +96,42 @@ class PulseConnection { std::condition_variable wait_cv_; }; +/// Holds the libpulse mainloop lock for a scope. Every pa_* call on a live context needs it. +class MainloopLock { +public: + explicit MainloopLock(pa_threaded_mainloop* loop) : loop_(loop) { + pa_threaded_mainloop_lock(this->loop_); + } + ~MainloopLock() { + pa_threaded_mainloop_unlock(this->loop_); + } + + MainloopLock(const MainloopLock&) = delete; + MainloopLock& operator=(const MainloopLock&) = delete; + +private: + pa_threaded_mainloop* loop_; +}; + +/// State shared by an asynchronous server query and its waiter; `done` is read unlocked. +struct PulseQuery { + PulseConnection* conn{nullptr}; + std::atomic done{false}; + bool failed{false}; + + /// Called from the query's own callback, on the mainloop thread, exactly once. + void finish(bool ok) { + this->failed = !ok; + this->done.store(true, std::memory_order_release); + this->conn->notify(); + } +}; + +/// Waits for one query, cancelling it on timeout so a late callback cannot touch a dead `query`. +/// The mainloop lock must NOT be held. +/// @return true if the query completed and the server answered it. +bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op); + /// An AudioSink that plays through a PulseAudio server; `-o pulse:` a sink name or empty. /// Lock order is mutex_ then the mainloop lock; no libpulse callback takes mutex_. class PulseAudioSink final : public AudioSink { From 4f65964497bb82037d1f5cd45ac7133387c3e5b8 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 7 Oct 2026 11:01:51 -0500 Subject: [PATCH 2/6] Capture through PipeWire and PulseAudio with --input --input pipewire[:] and --input pulse[:] open a capture stream on the sound server, and -l lists their sources. A named source is checked on every open and pinned to its device, so one that goes away is reopened with backoff instead of being swapped for the default. --- CMakeLists.txt | 4 +- src/audio_source.cpp | 110 +++++++++-- src/audio_source.h | 15 +- src/pipewire_source.cpp | 359 ++++++++++++++++++++++++++++++++++ src/pipewire_source.h | 85 ++++++++ src/pulse_source.cpp | 377 ++++++++++++++++++++++++++++++++++++ src/pulse_source.h | 76 ++++++++ tests/audio_source_test.cpp | 81 +++++++- tests/cli_test.cpp | 2 +- 9 files changed, 1084 insertions(+), 25 deletions(-) create mode 100644 src/pipewire_source.cpp create mode 100644 src/pipewire_source.h create mode 100644 src/pulse_source.cpp create mode 100644 src/pulse_source.h diff --git a/CMakeLists.txt b/CMakeLists.txt index b14647d..278e5a6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -304,13 +304,13 @@ if(SENDSPIN_CLI_PORTAUDIO_ENABLED) endif() if(SENDSPIN_CLI_PULSE_ENABLED) - target_sources(sendspin-cli-core PRIVATE src/pulse_sink.cpp) + target_sources(sendspin-cli-core PRIVATE src/pulse_sink.cpp src/pulse_source.cpp) target_link_libraries(sendspin-cli-core PUBLIC PkgConfig::PULSE) target_compile_definitions(sendspin-cli-core PUBLIC SENDSPIN_CLI_HAVE_PULSE) endif() if(SENDSPIN_CLI_PIPEWIRE_ENABLED) - target_sources(sendspin-cli-core PRIVATE src/pipewire_sink.cpp) + target_sources(sendspin-cli-core PRIVATE src/pipewire_sink.cpp src/pipewire_source.cpp) target_link_libraries(sendspin-cli-core PUBLIC PkgConfig::PIPEWIRE) target_compile_definitions(sendspin-cli-core PUBLIC SENDSPIN_CLI_HAVE_PIPEWIRE) endif() diff --git a/src/audio_source.cpp b/src/audio_source.cpp index 202da47..f96be92 100644 --- a/src/audio_source.cpp +++ b/src/audio_source.cpp @@ -21,6 +21,12 @@ #ifdef SENDSPIN_CLI_HAVE_ALSA #include "alsa_source.h" #endif +#ifdef SENDSPIN_CLI_HAVE_PIPEWIRE +#include "pipewire_source.h" +#endif +#ifdef SENDSPIN_CLI_HAVE_PULSE +#include "pulse_source.h" +#endif #include #include @@ -45,7 +51,18 @@ constexpr uint32_t READ_TIMEOUT_MS = 100; constexpr int64_t RATE_WINDOW_MS = 1000; /// Output backends --input knows by name but cannot capture through. -constexpr const char* UNSUPPORTED_BACKENDS[] = {"coreaudio", "portaudio", "pulse", "pipewire"}; +constexpr const char* UNSUPPORTED_BACKENDS[] = {"coreaudio", "portaudio"}; + +#ifdef SENDSPIN_CLI_HAVE_PULSE +constexpr bool HAVE_PULSE = true; +#else +constexpr bool HAVE_PULSE = false; +#endif +#ifdef SENDSPIN_CLI_HAVE_PIPEWIRE +constexpr bool HAVE_PIPEWIRE = true; +#else +constexpr bool HAVE_PIPEWIRE = false; +#endif } // namespace @@ -156,11 +173,30 @@ void SourceCapture::run_() { } std::string input_backend_list() { + std::string list = "null, tone"; #ifdef SENDSPIN_CLI_HAVE_ALSA - return "null, tone, alsa"; -#else - return "null, tone"; + list += ", alsa"; #endif + if (HAVE_PULSE) { + list += ", pulse"; + } + if (HAVE_PIPEWIRE) { + list += ", pipewire"; + } + return list; +} + +void settle_server_capture_format(StreamFormat& format) { + if (std::find(CAPTURE_BIT_DEPTHS.begin(), CAPTURE_BIT_DEPTHS.end(), format.bit_depth) == + CAPTURE_BIT_DEPTHS.end()) { + format.bit_depth = CAPTURE_BIT_DEPTHS.front(); + } + if (format.channels == 0) { + format.channels = 2; + } + if (format.sample_rate == 0) { + format.sample_rate = 48000; + } } bool resolve_input_spec(const std::string& spec, InputSpec& out, std::string& error) { @@ -201,18 +237,27 @@ bool resolve_input_spec(const std::string& spec, InputSpec& out, std::string& er #endif } - for (const char* name : UNSUPPORTED_BACKENDS) { - if (prefix != name) { - continue; - } - error = "capturing through the " + prefix + - " backend is not supported yet -- --input takes: " + input_backend_list(); + if (prefix == "pulse" || prefix == "pipewire") { + const bool pulse = prefix == "pulse"; + if (!(pulse ? HAVE_PULSE : HAVE_PIPEWIRE)) { + error = std::string("the ") + (pulse ? "PulseAudio" : "PipeWire") + + " backend is not in this build, so --input takes only: " + input_backend_list(); #ifdef SENDSPIN_CLI_HAVE_ALSA - if (prefix == "pulse" || prefix == "pipewire") { error += ". ALSA's plugin PCM of that name still works: --input alsa:" + prefix; - } #endif - return false; + return false; + } + out = {pulse ? SourceBackend::Pulse : SourceBackend::PipeWire, + rest == "default" ? std::string() : rest}; + return true; + } + + for (const char* name : UNSUPPORTED_BACKENDS) { + if (prefix == name) { + error = "capturing through the " + prefix + + " backend is not supported yet -- --input takes: " + input_backend_list(); + return false; + } } #ifdef SENDSPIN_CLI_HAVE_ALSA @@ -250,6 +295,16 @@ std::unique_ptr make_audio_source(const std::string& spec, StreamFo case SourceBackend::Alsa: #ifdef SENDSPIN_CLI_HAVE_ALSA source = std::make_unique(resolved.device); +#endif + break; + case SourceBackend::Pulse: +#ifdef SENDSPIN_CLI_HAVE_PULSE + source = std::make_unique(resolved.device); +#endif + break; + case SourceBackend::PipeWire: +#ifdef SENDSPIN_CLI_HAVE_PIPEWIRE + source = std::make_unique(resolved.device); #endif break; } @@ -284,9 +339,32 @@ void print_capture_devices(std::FILE* out) { "channels, S16_LE where the PCM takes it, else the nearest it does take. A\n" "plug-style PCM -- default, plughw: -- converts, so it takes the preferred\n" "format whatever the hardware behind it captures.\n"); -#else - std::fprintf(out, "\nThis build has no ALSA backend, so there is no sound card to capture\n" - "from here.\n"); +#elif !defined(SENDSPIN_CLI_HAVE_PULSE) && !defined(SENDSPIN_CLI_HAVE_PIPEWIRE) + std::fprintf(out, "\nThis build has no ALSA, PulseAudio or PipeWire backend, so there is no\n" + "sound card to capture from here.\n"); +#endif + +#ifdef SENDSPIN_CLI_HAVE_PULSE + std::fprintf(out, "\nPulseAudio sources on this host (--input pulse:):\n"); + PulseAudioSource::list_devices(out); + std::fprintf(out, + "\nThe name on each source's own line is what follows --input pulse:. --input\n" + "pulse with no source at all follows whichever one the server calls default.\n" + "A .monitor source captures what its sink is playing.\n"); +#endif + +#ifdef SENDSPIN_CLI_HAVE_PIPEWIRE + std::fprintf(out, "\nPipeWire audio source nodes on this host (--input pipewire:):\n"); + PipeWireAudioSource::list_devices(out); + std::fprintf(out, + "\nThe name on each node's own line is what follows --input pipewire:. --input\n" + "pipewire with no node at all captures wherever the graph routes a capture\n" + "stream.\n"); +#endif + +#if defined(SENDSPIN_CLI_HAVE_PULSE) || defined(SENDSPIN_CLI_HAVE_PIPEWIRE) + std::fprintf(out, "\nA sound server converts, so a capture through one runs at 48000 Hz, 2\n" + "channels, 16-bit whatever the hardware behind it captures.\n"); #endif } diff --git a/src/audio_source.h b/src/audio_source.h index 506b716..1b4f4d9 100644 --- a/src/audio_source.h +++ b/src/audio_source.h @@ -113,15 +113,17 @@ class SourceCapture final : public sendspin::SourceRoleListener { /// Which backend an --input spec named. enum class SourceBackend { - Null, ///< silence; needs no device - Tone, ///< a test tone; needs no device - Alsa, ///< an ALSA capture PCM, named by InputSpec::device + Null, ///< silence; needs no device + Tone, ///< a test tone; needs no device + Alsa, ///< an ALSA capture PCM, named by InputSpec::device + Pulse, ///< a PulseAudio source, or the server's default with no device + PipeWire, ///< a PipeWire source node, or the graph's default with no device }; /// An --input spec resolved into a backend and the device to hand it. struct InputSpec { SourceBackend backend{SourceBackend::Null}; - std::string device; ///< empty for the device-less sources + std::string device; ///< empty for the device-less sources and a sound server's default }; /// Resolves an --input spec without opening anything, by -o's rules: a bare backend name, @@ -129,12 +131,15 @@ struct InputSpec { /// @return true if `spec` named something this build can capture from. bool resolve_input_spec(const std::string& spec, InputSpec& out, std::string& error); -/// The backend prefixes --input has in this build, e.g. "null, tone, alsa". +/// The backend prefixes --input has in this build, e.g. "null, tone, alsa, pulse". std::string input_backend_list(); /// True if a bare `--input ` reaches the ALSA PCM of that name. bool input_pcm_is_reachable(const std::string& pcm); +/// Settles `format` for a sound server, which converts: only a format no stream can carry changes. +void settle_server_capture_format(StreamFormat& format); + /// Builds the source named by --input and negotiates its capture format. /// @param format In: the format to prefer. Out: the one the device will capture. /// @return The source, or nullptr with `error` set. diff --git a/src/pipewire_source.cpp b/src/pipewire_source.cpp new file mode 100644 index 0000000..abb53f5 --- /dev/null +++ b/src/pipewire_source.cpp @@ -0,0 +1,359 @@ +// Copyright 2026 sendspin-cpp-cli Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "pipewire_source.h" + +#include "log.h" + +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace sendspin_cli { + +using sendspin::LogLevel; + +static constexpr const char* LOG_TAG = LOG_TAG_AUDIO; + +namespace { + +constexpr const char* PIPEWIRE_APP_NAME = "sendspin-cli"; + +/// The name the capture stream's node carries in the graph, apart from the playback stream's. +constexpr const char* PIPEWIRE_NODE_NAME = "sendspin-cli-input"; + +/// Quantum asked of the graph; one source-role chunk. +constexpr uint32_t QUANTUM_MS = 20; + +/// Ring size; whole quanta are dropped once a stalled reader has let it fill. +constexpr uint32_t RING_MS = 1000; + +constexpr int64_t US_PER_S = 1000000; + +/// SPA's packed little-endian format for a capture bit depth. +spa_audio_format spa_format_for(uint8_t bit_depth) { + switch (bit_depth) { + case 16: + return SPA_AUDIO_FORMAT_S16_LE; + case 24: + return SPA_AUDIO_FORMAT_S24_LE; + case 32: + return SPA_AUDIO_FORMAT_S32_LE; + default: + return SPA_AUDIO_FORMAT_UNKNOWN; + } +} + +/// Lists the graph's capture nodes, virtual ones included. +bool list_source_nodes(std::vector& out, std::string& error) { + return pipewire_list_nodes({"Audio/Source", "Audio/Source/Virtual"}, out, error); +} + +} // namespace + +PipeWireAudioSource::PipeWireAudioSource(std::string device) : device_(std::move(device)) {} + +PipeWireAudioSource::~PipeWireAudioSource() { + this->close(); +} + +std::string PipeWireAudioSource::name() const { + return this->device_.empty() ? "pipewire" : "pipewire:" + this->device_; +} + +void PipeWireAudioSource::list_devices(std::FILE* out) { + const PipeWireGuard guard; + + std::vector nodes; + std::string error; + if (!list_source_nodes(nodes, error)) { + std::fprintf(out, " (%s)\n", error.c_str()); + return; + } + if (nodes.empty()) { + std::fprintf(out, " (this graph has no audio source nodes)\n"); + return; + } + for (const PipeWireNode& node : nodes) { + std::fprintf(out, " %s\n", node.name.c_str()); + if (!node.description.empty()) { + std::fprintf(out, " %s\n", node.description.c_str()); + } + } +} + +bool PipeWireAudioSource::find_node_(std::string& error) const { + std::vector nodes; + if (!list_source_nodes(nodes, error)) { + return false; + } + const bool found = + this->device_.empty() || + std::any_of(nodes.begin(), nodes.end(), + [this](const PipeWireNode& node) { return node.name == this->device_; }); + if (!found) { + error = "--input " + this->name() + ": this graph has no audio source node by that name"; + } + return found; +} + +bool PipeWireAudioSource::negotiate(StreamFormat& format, std::string& error) { + if (!this->find_node_(error)) { + error += " -- run with -l to list this host's capture devices"; + return false; + } + settle_server_capture_format(format); + return true; +} + +bool PipeWireAudioSource::open(const StreamFormat& format) { + this->close(); + + const spa_audio_format spa_format = spa_format_for(format.bit_depth); + if (spa_format == SPA_AUDIO_FORMAT_UNKNOWN || format.channels == 0 || format.sample_rate == 0) { + cli_log(LogLevel::ERROR, "pipewire: input cannot capture %u Hz / %u ch / %u-bit", + format.sample_rate, format.channels, format.bit_depth); + return false; + } + // A named node that has gone would otherwise be swapped for the default by the session manager. + std::string error; + if (!this->device_.empty() && !this->find_node_(error)) { + cli_log(LogLevel::ERROR, "pipewire: %s", error.c_str()); + return false; + } + + this->loop_ = pw_thread_loop_new("sendspin-pw-in", nullptr); + if (this->loop_ == nullptr || pw_thread_loop_start(this->loop_) < 0) { + cli_log(LogLevel::ERROR, "pipewire: cannot start a loop for input '%s'", + this->name().c_str()); + this->close(); + return false; + } + + this->rate_ = format.sample_rate; + this->bytes_per_frame_ = + static_cast(format.channels) * (static_cast(format.bit_depth) / 8U); + // The spare byte the ring keeps to tell full from empty. + this->ring_.reset((static_cast(format.sample_rate) * RING_MS / 1000U * + this->bytes_per_frame_) + + 1); + + pw_properties* props = pw_properties_new( + PW_KEY_MEDIA_TYPE, "Audio", PW_KEY_MEDIA_CATEGORY, "Capture", PW_KEY_APP_NAME, + PIPEWIRE_APP_NAME, PW_KEY_NODE_NAME, PIPEWIRE_NODE_NAME, nullptr); + if (props == nullptr) { + cli_log(LogLevel::ERROR, "pipewire: cannot describe the input stream to the graph"); + this->close(); + return false; + } + if (!this->device_.empty()) { + pw_properties_set(props, PW_KEY_TARGET_OBJECT, this->device_.c_str()); + // Unplugging the node ends the stream, so it is reopened rather than moved elsewhere. + pw_properties_set(props, PW_KEY_NODE_DONT_RECONNECT, "true"); + } + const std::string latency = std::to_string(format.sample_rate * QUANTUM_MS / 1000U) + "/" + + std::to_string(format.sample_rate); + pw_properties_set(props, PW_KEY_NODE_LATENCY, latency.c_str()); + + uint8_t pod_storage[1024]; + spa_pod_builder builder = SPA_POD_BUILDER_INIT(pod_storage, sizeof(pod_storage)); + spa_audio_info_raw info{}; + info.format = spa_format; + info.rate = format.sample_rate; + info.channels = format.channels; + if (format.channels == 1) { + info.position[0] = SPA_AUDIO_CHANNEL_MONO; + } else if (format.channels == 2) { + info.position[0] = SPA_AUDIO_CHANNEL_FL; + info.position[1] = SPA_AUDIO_CHANNEL_FR; + } + const spa_pod* params[1]; + params[0] = spa_format_audio_raw_build(&builder, SPA_PARAM_EnumFormat, &info); + + // Static: pw_stream_new_simple() keeps the pointer. Zeroed then assigned: the struct grows. + static constexpr pw_stream_events events = [] { + pw_stream_events table{}; + table.version = PW_VERSION_STREAM_EVENTS; + table.state_changed = &PipeWireAudioSource::stream_state_cb; + table.process = &PipeWireAudioSource::stream_process_cb; + return table; + }(); + + int err = -ENOMEM; + { + const LoopLock lock(this->loop_); + // pw_stream_new_simple() consumes props even on failure. + this->stream_ = pw_stream_new_simple(pw_thread_loop_get_loop(this->loop_), + PIPEWIRE_APP_NAME, props, &events, this); + if (this->stream_ != nullptr) { + const auto flags = static_cast( + PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS); + err = pw_stream_connect(this->stream_, PW_DIRECTION_INPUT, PW_ID_ANY, flags, params, 1); + } + if (err >= 0) { + // PAUSED counts: the graph has taken the stream and starts it when a source is linked. + timespec deadline{}; + pw_thread_loop_get_time(this->loop_, &deadline, + static_cast(PIPEWIRE_RECOVERY_TIMEOUT_MS) * 1000 * 1000); + while (true) { + const auto state = static_cast(this->stream_state_.load()); + if (state == PW_STREAM_STATE_STREAMING || state == PW_STREAM_STATE_PAUSED || + state == PW_STREAM_STATE_ERROR) { + break; + } + if (pw_thread_loop_timed_wait_full(this->loop_, &deadline) != 0) { + break; + } + } + } + } + + const auto state = static_cast(this->stream_state_.load()); + if (err < 0) { + cli_log(LogLevel::ERROR, "pipewire: cannot capture from '%s': %s", this->name().c_str(), + spa_strerror(err)); + } else if (state == PW_STREAM_STATE_ERROR) { + err = -EIO; // stream_state_cb() has said why + } else if (state != PW_STREAM_STATE_STREAMING && state != PW_STREAM_STATE_PAUSED) { + err = -ETIMEDOUT; + cli_log(LogLevel::ERROR, "pipewire: input '%s' would not start at %u Hz, %u ch, %u-bit", + this->name().c_str(), format.sample_rate, format.channels, format.bit_depth); + } + if (err < 0) { + this->close(); + return false; + } + + cli_log(LogLevel::INFO, "pipewire: input '%s' open at %u Hz, %u ch, %u-bit", + this->name().c_str(), format.sample_rate, format.channels, format.bit_depth); + return true; +} + +void PipeWireAudioSource::stream_state_cb(void* userdata, enum pw_stream_state /*old*/, + enum pw_stream_state state, const char* error) { + auto* self = static_cast(userdata); + self->stream_state_.store(static_cast(state)); + if (state == PW_STREAM_STATE_ERROR) { + cli_log(LogLevel::WARN, "pipewire: input '%s' failed: %s", self->name().c_str(), + (error != nullptr) ? error : "(no reason given)"); + } + if (state == PW_STREAM_STATE_ERROR || state == PW_STREAM_STATE_UNCONNECTED) { + self->lost_.store(true); + self->data_ready_.notify_all(); + } + pw_thread_loop_signal(self->loop_, false); +} + +void PipeWireAudioSource::stream_process_cb(void* userdata) { + auto* self = static_cast(userdata); + + // Safe unlocked: pw_stream_disconnect() waits for this callback to return. + pw_buffer* buffer = pw_stream_dequeue_buffer(self->stream_); + if (buffer == nullptr) { + return; + } + const spa_data& data = buffer->buffer->datas[0]; + const size_t stride = self->bytes_per_frame_; + if (data.data != nullptr && stride != 0) { + const uint32_t offset = std::min(data.chunk->offset, data.maxsize); + const size_t frames = std::min(data.chunk->size, data.maxsize - offset) / stride; + // All or nothing, so the frames in the ring stay contiguous within a quantum. + if (frames != 0 && self->ring_.free_space() >= frames * stride) { + pw_time time{}; + if (pw_stream_get_time_n(self->stream_, &time, sizeof(time)) == 0 && + time.rate.denom != 0) { + // `delay` is the newest frame's age; the first is older by this buffer and by + // what the converter still holds. + const double age_s = (static_cast(time.delay) * + static_cast(time.rate.num) / + static_cast(time.rate.denom)) + + (static_cast(time.buffered + frames) / + static_cast(self->rate_)); + // `now` is CLOCK_MONOTONIC, which is what steady_clock reads here. + const int64_t first_us = (time.now / 1000) - static_cast(age_s * 1e6); + self->origin_us_.store(first_us - static_cast(self->frames_written_) * + US_PER_S / self->rate_); + } + self->ring_.write(static_cast(data.data) + offset, frames * stride); + self->frames_written_ += frames; + } + } + pw_stream_queue_buffer(self->stream_, buffer); + + // Notified without the mutex; a missed wakeup costs at most one quantum. + self->data_ready_.notify_one(); +} + +int PipeWireAudioSource::read(uint8_t* data, size_t length, uint32_t timeout_ms, + int64_t& capture_time_us) { + if (this->stream_ == nullptr) { + return -1; + } + { + std::unique_lock lock(this->wait_mutex_); + this->data_ready_.wait_for(lock, std::chrono::milliseconds(timeout_ms), [this] { + return this->lost_.load() || this->ring_.available() >= this->bytes_per_frame_; + }); + } + if (this->lost_.load()) { + return -1; + } + + size_t bytes = std::min(this->ring_.available(), length); + bytes -= bytes % this->bytes_per_frame_; + if (bytes == 0) { + return 0; + } + this->ring_.read(data, bytes); + + const int64_t origin_us = this->origin_us_.load(); + capture_time_us = + (origin_us == 0) + ? 0 + : origin_us + static_cast(this->frames_read_) * US_PER_S / this->rate_; + this->frames_read_ += bytes / this->bytes_per_frame_; + return static_cast(bytes); +} + +void PipeWireAudioSource::close() { + if (this->stream_ != nullptr) { + const LoopLock lock(this->loop_); + // Disconnect before destroy, under the loop lock: disconnect waits out process(). + pw_stream_disconnect(this->stream_); + pw_stream_destroy(this->stream_); + this->stream_ = nullptr; + } + // The loop goes too: a restarted daemon took the old connection with it. + if (this->loop_ != nullptr) { + pw_thread_loop_destroy(this->loop_); + this->loop_ = nullptr; + } + + this->ring_.reset(0); + this->rate_ = 0; + this->bytes_per_frame_ = 0; + this->frames_written_ = 0; + this->frames_read_ = 0; + this->origin_us_.store(0); + this->lost_.store(false); + this->stream_state_.store(PW_STREAM_STATE_UNCONNECTED); +} + +} // namespace sendspin_cli diff --git a/src/pipewire_source.h b/src/pipewire_source.h new file mode 100644 index 0000000..e5c81c3 --- /dev/null +++ b/src/pipewire_source.h @@ -0,0 +1,85 @@ +// Copyright 2026 sendspin-cpp-cli Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/// AudioSource over a libpipewire capture stream, timestamped from the graph's reported delay. + +#pragma once + +#include "audio_source.h" +#include "pcm_ring.h" +#include "pipewire_sink.h" + +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace sendspin_cli { + +/// An AudioSource that captures from a PipeWire graph; `--input pipewire:` a node name or empty. +/// process() writes the ring and reads the format fields unlocked: change them only while closed. +class PipeWireAudioSource final : public AudioSource { +public: + /// @param device Source node name, or empty for the graph's default. + explicit PipeWireAudioSource(std::string device); + ~PipeWireAudioSource() override; + + std::string name() const override; + bool negotiate(StreamFormat& format, std::string& error) override; + bool open(const StreamFormat& format) override; + int read(uint8_t* data, size_t length, uint32_t timeout_ms, int64_t& capture_time_us) override; + void close() override; + + /// Prints this graph's audio source nodes. + static void list_devices(std::FILE* out); + +private: + static void stream_state_cb(void* userdata, enum pw_stream_state old, + enum pw_stream_state state, const char* error); + static void stream_process_cb(void* userdata); + + /// Checks that a daemon answers and has the named node. + bool find_node_(std::string& error) const; + + /// Declared first so it is destroyed last, after the stream and its loop. + PipeWireGuard pw_; + + std::string device_; ///< empty for the graph's default + + pw_thread_loop* loop_{nullptr}; + pw_stream* stream_{nullptr}; + PcmRingBuffer ring_; + uint32_t rate_{0}; + size_t bytes_per_frame_{0}; + + uint64_t frames_written_{0}; ///< process() only + uint64_t frames_read_{0}; ///< read() only + /// Steady-clock time of this stream's frame 0 as process() last measured it; 0 until it has. + std::atomic origin_us_{0}; + + /// Set when the graph errors or drops the stream. + std::atomic lost_{false}; + std::atomic stream_state_{PW_STREAM_STATE_UNCONNECTED}; + + std::mutex wait_mutex_; + /// Signalled by process(), without wait_mutex_, once it has added to the ring. + std::condition_variable data_ready_; +}; + +} // namespace sendspin_cli diff --git a/src/pulse_source.cpp b/src/pulse_source.cpp new file mode 100644 index 0000000..6b629c6 --- /dev/null +++ b/src/pulse_source.cpp @@ -0,0 +1,377 @@ +// Copyright 2026 sendspin-cpp-cli Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "pulse_source.h" + +#include "log.h" + +#include +#include +#include +#include +#include + +namespace sendspin_cli { + +using sendspin::LogLevel; + +static constexpr const char* LOG_TAG = LOG_TAG_AUDIO; + +namespace { + +/// The stream name the server shows in its mixer. +constexpr const char* PULSE_STREAM_NAME = "sendspin-cli input"; + +/// Fragment asked of the server; one source-role chunk. +constexpr pa_usec_t FRAGMENT_US = 20000; + +constexpr int64_t US_PER_S = 1000000; + +int64_t now_us() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + +/// PulseAudio's packed little-endian format for a capture bit depth. +pa_sample_format_t pulse_format_for(uint8_t bit_depth) { + switch (bit_depth) { + case 16: + return PA_SAMPLE_S16LE; + case 24: + return PA_SAMPLE_S24LE; + case 32: + return PA_SAMPLE_S32LE; + default: + return PA_SAMPLE_INVALID; + } +} + +/// One source as the server describes it. +struct PulseSourceInfo { + std::string name; + std::string description; + pa_sample_spec spec{}; +}; + +struct SourceListQuery : PulseQuery { + std::vector sources; +}; + +struct DefaultSourceQuery : PulseQuery { + std::string name; +}; + +void source_info_cb(pa_context* /*context*/, const pa_source_info* info, int eol, void* userdata) { + auto* query = static_cast(userdata); + if (eol != 0) { + query->finish(eol > 0); + return; + } + PulseSourceInfo entry; + entry.name = (info->name != nullptr) ? info->name : ""; + entry.description = (info->description != nullptr) ? info->description : ""; + entry.spec = info->sample_spec; + query->sources.push_back(std::move(entry)); +} + +void server_info_cb(pa_context* /*context*/, const pa_server_info* info, void* userdata) { + auto* query = static_cast(userdata); + if (info != nullptr && info->default_source_name != nullptr) { + query->name = info->default_source_name; + } + query->finish(info != nullptr); +} + +/// Asks the server for every source it has, monitors included. The mainloop lock must NOT be held. +bool list_sources(PulseConnection& conn, std::vector& out) { + SourceListQuery query; + query.conn = &conn; + pa_operation* op = nullptr; + { + const MainloopLock lock(conn.mainloop()); + op = pa_context_get_source_info_list(conn.context(), source_info_cb, &query); + } + if (!await_query(conn, query, op)) { + return false; + } + out = std::move(query.sources); + return true; +} + +/// Asks the server which source it calls default. Empty when it will not say. Lock must NOT be +/// held. +std::string default_source_name(PulseConnection& conn) { + DefaultSourceQuery query; + query.conn = &conn; + pa_operation* op = nullptr; + { + const MainloopLock lock(conn.mainloop()); + op = pa_context_get_server_info(conn.context(), server_info_cb, &query); + } + return await_query(conn, query, op) ? query.name : ""; +} + +} // namespace + +PulseAudioSource::PulseAudioSource(std::string device) : device_(std::move(device)) {} + +PulseAudioSource::~PulseAudioSource() { + this->close(); +} + +std::string PulseAudioSource::name() const { + return this->device_.empty() ? "pulse" : "pulse:" + this->device_; +} + +void PulseAudioSource::list_devices(std::FILE* out) { + PulseConnection conn; + std::string error; + if (!conn.connect(error)) { + std::fprintf(out, " (%s)\n", error.c_str()); + return; + } + + std::vector sources; + if (!list_sources(conn, sources)) { + std::fprintf(out, " (cannot list the sources on %s)\n", conn.server_name().c_str()); + return; + } + if (sources.empty()) { + std::fprintf(out, " (%s has no sources)\n", conn.server_name().c_str()); + return; + } + + const std::string fallback = default_source_name(conn); + for (const PulseSourceInfo& source : sources) { + std::fprintf(out, " %s%s\n", source.name.c_str(), + (source.name == fallback) ? " (server default)" : ""); + std::fprintf(out, " %s\n", source.description.c_str()); + char spec[PA_SAMPLE_SPEC_SNPRINT_MAX]; + pa_sample_spec_snprint(spec, sizeof(spec), &source.spec); + std::fprintf(out, " running at %s\n", spec); + } +} + +bool PulseAudioSource::find_source_(std::string& error) { + std::vector sources; + if (!list_sources(this->conn_, sources)) { + error = "cannot list the sources on " + this->conn_.server_name(); + return false; + } + const bool found = + std::any_of(sources.begin(), sources.end(), + [this](const PulseSourceInfo& source) { return source.name == this->device_; }); + if (!found) { + error = "--input " + this->name() + ": " + this->conn_.server_name() + + " has no source by that name"; + } + return found; +} + +bool PulseAudioSource::negotiate(StreamFormat& format, std::string& error) { + if (!this->conn_.connect(error) || (!this->device_.empty() && !this->find_source_(error))) { + error += " -- run with -l to list this host's capture devices"; + return false; + } + settle_server_capture_format(format); + return true; +} + +bool PulseAudioSource::open(const StreamFormat& format) { + this->close(); + + pa_sample_spec spec{}; + spec.format = pulse_format_for(format.bit_depth); + spec.rate = format.sample_rate; + spec.channels = format.channels; + if (pa_sample_spec_valid(&spec) == 0) { + cli_log(LogLevel::ERROR, "pulse: input cannot capture %u Hz / %u ch / %u-bit", + format.sample_rate, format.channels, format.bit_depth); + return false; + } + if (!this->conn_.ready()) { + std::string error; + if (!this->conn_.connect(error, PULSE_RECOVERY_TIMEOUT_MS)) { + cli_log(LogLevel::ERROR, "pulse: %s", error.c_str()); + return false; + } + } + // pipewire-pulse records from the default when the named source is missing. + std::string error; + if (!this->device_.empty() && !this->find_source_(error)) { + cli_log(LogLevel::ERROR, "pulse: %s", error.c_str()); + return false; + } + + this->rate_ = format.sample_rate; + this->bytes_per_frame_ = pa_frame_size(&spec); + + pa_buffer_attr attr{}; + attr.maxlength = static_cast(-1); + attr.tlength = static_cast(-1); + attr.prebuf = static_cast(-1); + attr.minreq = static_cast(-1); + attr.fragsize = static_cast(pa_usec_to_bytes(FRAGMENT_US, &spec)); + + int flags = + PA_STREAM_ADJUST_LATENCY | PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_AUTO_TIMING_UPDATE; + if (!this->device_.empty()) { + // Unplugging the source ends the stream, so it is reopened rather than moved elsewhere. + flags |= PA_STREAM_DONT_MOVE; + } + const char* device = this->device_.empty() ? nullptr : this->device_.c_str(); + + int err = -1; + { + const MainloopLock ml(this->conn_.mainloop()); + this->stream_ = pa_stream_new(this->conn_.context(), PULSE_STREAM_NAME, &spec, nullptr); + if (this->stream_ != nullptr) { + pa_stream_set_state_callback(this->stream_, &PulseAudioSource::stream_state_cb, this); + pa_stream_set_read_callback(this->stream_, &PulseAudioSource::stream_read_cb, this); + err = pa_stream_connect_record(this->stream_, device, &attr, + static_cast(flags)); + } + if (err < 0) { + cli_log(LogLevel::ERROR, "pulse: cannot capture from '%s': %s", this->name().c_str(), + pa_strerror(pa_context_errno(this->conn_.context()))); + } + } + if (err < 0) { + this->close(); + return false; + } + + const bool settled = this->conn_.wait_for( + [this] { + const auto state = static_cast(this->stream_state_.load()); + return state == PA_STREAM_READY || PA_STREAM_IS_GOOD(state) == 0; + }, + PULSE_RECOVERY_TIMEOUT_MS); + if (!settled || this->stream_state_.load() != PA_STREAM_READY) { + cli_log(LogLevel::ERROR, "pulse: input '%s' would not start at %u Hz, %u ch, %u-bit -- %s", + this->name().c_str(), format.sample_rate, format.channels, format.bit_depth, + settled ? "the server refused it" : "the server did not answer"); + this->close(); + return false; + } + + cli_log(LogLevel::INFO, "pulse: input '%s' open on %s at %u Hz, %u ch, %u-bit", + this->name().c_str(), this->conn_.server_name().c_str(), format.sample_rate, + format.channels, format.bit_depth); + return true; +} + +void PulseAudioSource::stream_state_cb(pa_stream* stream, void* userdata) { + auto* self = static_cast(userdata); + const pa_stream_state_t state = pa_stream_get_state(stream); + self->stream_state_.store(static_cast(state)); + if (state == PA_STREAM_FAILED || state == PA_STREAM_TERMINATED) { + self->lost_.store(true); + } + self->conn_.notify(); +} + +void PulseAudioSource::stream_read_cb(pa_stream* /*stream*/, size_t /*nbytes*/, void* userdata) { + auto* self = static_cast(userdata); + self->readable_.store(true); + self->conn_.notify(); +} + +int PulseAudioSource::drain_(uint8_t* data, size_t length, int64_t& capture_time_us) { + if (this->lost_.load() || !this->conn_.ready()) { + return -1; + } + const MainloopLock ml(this->conn_.mainloop()); + + // The server's measure of how old the sample at the read index is. + pa_usec_t latency_us = 0; + int negative = 0; + const bool timed = + pa_stream_get_latency(this->stream_, &latency_us, &negative) == 0 && negative == 0; + // The read index only moves with a whole fragment, so the part already returned is added back. + const int64_t first_us = now_us() - static_cast(latency_us) + + static_cast(this->fragment_offset_ / this->bytes_per_frame_) * + US_PER_S / this->rate_; + + size_t done = 0; + while (length - done >= this->bytes_per_frame_) { + const void* fragment = nullptr; + size_t size = 0; + if (pa_stream_peek(this->stream_, &fragment, &size) < 0) { + return -1; + } + if (size == 0) { + break; + } + if (fragment == nullptr) { + // A hole: skipped, and the read ends here so no timestamp spans it. + pa_stream_drop(this->stream_); + break; + } + size_t chunk = std::min(size - this->fragment_offset_, length - done); + chunk -= chunk % this->bytes_per_frame_; + std::memcpy(data + done, static_cast(fragment) + this->fragment_offset_, + chunk); + done += chunk; + this->fragment_offset_ += chunk; + if (size - this->fragment_offset_ < this->bytes_per_frame_) { + pa_stream_drop(this->stream_); + this->fragment_offset_ = 0; + } + } + if (done > 0) { + capture_time_us = timed ? first_us : 0; + } + return static_cast(done); +} + +int PulseAudioSource::read(uint8_t* data, size_t length, uint32_t timeout_ms, + int64_t& capture_time_us) { + if (this->stream_ == nullptr) { + return -1; + } + // Cleared before the server is asked, so a fragment landing in between still ends the wait. + this->readable_.store(false); + int bytes = this->drain_(data, length, capture_time_us); + if (bytes == 0 && this->conn_.wait_for( + [this] { + return this->readable_.load() || this->lost_.load() || + !this->conn_.ready(); + }, + static_cast(timeout_ms))) { + bytes = this->drain_(data, length, capture_time_us); + } + return bytes; +} + +void PulseAudioSource::close() { + if (this->stream_ != nullptr && this->conn_.mainloop() != nullptr) { + const MainloopLock ml(this->conn_.mainloop()); + // Cleared before the disconnect, whose own state change must not reach a dying stream. + pa_stream_set_state_callback(this->stream_, nullptr, nullptr); + pa_stream_set_read_callback(this->stream_, nullptr, nullptr); + pa_stream_disconnect(this->stream_); + pa_stream_unref(this->stream_); + } + this->stream_ = nullptr; + this->rate_ = 0; + this->bytes_per_frame_ = 0; + this->fragment_offset_ = 0; + this->readable_.store(false); + this->lost_.store(false); + this->stream_state_.store(PA_STREAM_UNCONNECTED); +} + +} // namespace sendspin_cli diff --git a/src/pulse_source.h b/src/pulse_source.h new file mode 100644 index 0000000..f76aae7 --- /dev/null +++ b/src/pulse_source.h @@ -0,0 +1,76 @@ +// Copyright 2026 sendspin-cpp-cli Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +/// AudioSource over a libpulse record stream, timestamped from the server-reported latency. + +#pragma once + +#include "audio_source.h" +#include "pulse_sink.h" + +#include + +#include +#include +#include +#include +#include + +namespace sendspin_cli { + +/// An AudioSource that captures from a PulseAudio server; `--input pulse:` a source name or empty. +class PulseAudioSource final : public AudioSource { +public: + /// @param device Source name, or empty for the server's default. + explicit PulseAudioSource(std::string device); + ~PulseAudioSource() override; + + std::string name() const override; + bool negotiate(StreamFormat& format, std::string& error) override; + bool open(const StreamFormat& format) override; + int read(uint8_t* data, size_t length, uint32_t timeout_ms, int64_t& capture_time_us) override; + void close() override; + + /// Prints this server's sources, with the default marked. + static void list_devices(std::FILE* out); + +private: + static void stream_state_cb(pa_stream* stream, void* userdata); + static void stream_read_cb(pa_stream* stream, size_t nbytes, void* userdata); + + /// Checks that the connected server has the named source. + bool find_source_(std::string& error); + + /// Copies out what the server has queued, without waiting. + /// @return Bytes copied, 0 if nothing is queued, negative once the stream is lost. + int drain_(uint8_t* data, size_t length, int64_t& capture_time_us); + + /// Declared first so it is destroyed last, after the stream is freed. + PulseConnection conn_; + + std::string device_; ///< empty for the server's default + + pa_stream* stream_{nullptr}; + uint32_t rate_{0}; + size_t bytes_per_frame_{0}; + size_t fragment_offset_{0}; ///< bytes of the peeked fragment already returned + + /// Set by the read callback; cleared by read() before it asks the server. + std::atomic readable_{false}; + /// Set when the server kills the stream. + std::atomic lost_{false}; + std::atomic stream_state_{PA_STREAM_UNCONNECTED}; +}; + +} // namespace sendspin_cli diff --git a/tests/audio_source_test.cpp b/tests/audio_source_test.cpp index 2552217..ccbdf1f 100644 --- a/tests/audio_source_test.cpp +++ b/tests/audio_source_test.cpp @@ -17,6 +17,7 @@ #include "audio_source.h" #include "null_source.h" +#include "scoped_env.h" #include @@ -146,13 +147,91 @@ TEST(ResolveInputSpec, DeviceLessNamesRefuseADevice) { } TEST(ResolveInputSpec, OutputOnlyBackendsAreRefusedByName) { - for (const char* spec : {"pulse", "pulse:mic", "pipewire:node", "coreaudio", "portaudio:1"}) { + for (const char* spec : {"coreaudio", "coreaudio:1", "portaudio:1"}) { const std::string error = rejected(spec); EXPECT_NE(error.find("not supported yet"), std::string::npos) << spec; EXPECT_NE(error.find(input_backend_list()), std::string::npos) << spec; } } +#ifdef SENDSPIN_CLI_HAVE_PULSE +TEST(ResolveInputSpec, PulseTakesASourceOrTheServerDefault) { + for (const char* spec : {"pulse", "pulse:", "pulse:default"}) { + EXPECT_EQ(resolved(spec).backend, SourceBackend::Pulse) << spec; + EXPECT_TRUE(resolved(spec).device.empty()) << spec; + } + EXPECT_EQ(resolved("pulse:alsa_input.usb-mic.analog-stereo").device, + "alsa_input.usb-mic.analog-stereo"); + // Split on the first colon only: the rest is the source's name, colons and all. + EXPECT_EQ(resolved("pulse:a:b").device, "a:b"); + EXPECT_NE(input_backend_list().find("pulse"), std::string::npos); +} + +TEST(MakeAudioSource, AnUnreachablePulseServerIsRefusedNamingTheListing) { + const ScopedEnv server("PULSE_SERVER", "unix:/nonexistent/sendspin-cli-test"); + StreamFormat format = STEREO_16; + std::string error; + EXPECT_EQ(make_audio_source("pulse", format, error), nullptr); + EXPECT_NE(error.find("PulseAudio"), std::string::npos) << error; + EXPECT_NE(error.find("-l"), std::string::npos) << error; +} +#else +TEST(ResolveInputSpec, WithoutPulseItsNameIsNotInThisBuild) { + for (const char* spec : {"pulse", "pulse:mic"}) { + const std::string error = rejected(spec); + EXPECT_NE(error.find("PulseAudio backend is not in this build"), std::string::npos) << spec; + EXPECT_NE(error.find(input_backend_list()), std::string::npos) << spec; + } +} +#endif + +#ifdef SENDSPIN_CLI_HAVE_PIPEWIRE +TEST(ResolveInputSpec, PipeWireTakesANodeOrTheGraphDefault) { + for (const char* spec : {"pipewire", "pipewire:", "pipewire:default"}) { + EXPECT_EQ(resolved(spec).backend, SourceBackend::PipeWire) << spec; + EXPECT_TRUE(resolved(spec).device.empty()) << spec; + } + EXPECT_EQ(resolved("pipewire:alsa_input.pci-0000_00_1f.3.analog-stereo").device, + "alsa_input.pci-0000_00_1f.3.analog-stereo"); + EXPECT_NE(input_backend_list().find("pipewire"), std::string::npos); +} + +TEST(MakeAudioSource, AnUnreachablePipeWireDaemonIsRefusedNamingTheListing) { + const ScopedEnv remote("PIPEWIRE_REMOTE", "/nonexistent/sendspin-cli-test"); + StreamFormat format = STEREO_16; + std::string error; + EXPECT_EQ(make_audio_source("pipewire", format, error), nullptr); + EXPECT_NE(error.find("PipeWire"), std::string::npos) << error; + EXPECT_NE(error.find("-l"), std::string::npos) << error; +} +#else +TEST(ResolveInputSpec, WithoutPipeWireItsNameIsNotInThisBuild) { + for (const char* spec : {"pipewire", "pipewire:node"}) { + const std::string error = rejected(spec); + EXPECT_NE(error.find("PipeWire backend is not in this build"), std::string::npos) << spec; + EXPECT_NE(error.find(input_backend_list()), std::string::npos) << spec; + } +} +#endif + +TEST(SettleServerCaptureFormat, KeepsWhatAStreamCanCarry) { + for (const uint8_t depth : CAPTURE_BIT_DEPTHS) { + StreamFormat format{44100, 1, depth}; + settle_server_capture_format(format); + EXPECT_EQ(format.sample_rate, 44100U); + EXPECT_EQ(format.channels, 1); + EXPECT_EQ(format.bit_depth, depth); + } +} + +TEST(SettleServerCaptureFormat, ReplacesWhatItCannot) { + StreamFormat format{0, 0, 8}; + settle_server_capture_format(format); + EXPECT_EQ(format.sample_rate, 48000U); + EXPECT_EQ(format.channels, 2); + EXPECT_EQ(format.bit_depth, 16); +} + #ifdef SENDSPIN_CLI_HAVE_ALSA TEST(ResolveInputSpec, BareNamesAreAlsaPcmsColonsAndAll) { EXPECT_EQ(resolved("default").backend, SourceBackend::Alsa); diff --git a/tests/cli_test.cpp b/tests/cli_test.cpp index a117280..66a3325 100644 --- a/tests/cli_test.cpp +++ b/tests/cli_test.cpp @@ -321,7 +321,7 @@ TEST(ParseOptions, InputTakesADeviceLessSource) { TEST(ParseOptions, ABadInputSpecIsRefusedAtTheFlag) { Parse device_on_tone({"--input", "tone:hw:0"}); - Parse unsupported({"--input", "pulse:mic"}); + Parse unsupported({"--input", "portaudio:1"}); Parse empty({"--input", ""}); EXPECT_FALSE(device_on_tone.ok()); From c6927e6ea9c43ccf8c7d8d33f6780f8df44db077 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 7 Oct 2026 11:01:51 -0500 Subject: [PATCH 3/6] Document capturing through PulseAudio and PipeWire --- README.md | 2 +- docs/wiki/Advanced-Usage.md | 40 +++++++++++++++++++---------- docs/wiki/Configuration.md | 2 +- packaging/sendspin-cli.conf.example | 1 + 4 files changed, 30 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index eced85c..70614aa 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,7 @@ sendspin-cli -n den --input default | `-n, --name ` | The friendly name a server displays. Defaults to this host's name. | | `-o, --output ` | Which sound card to play through. `-l` lists this host's devices and what they accept. | | `--audio-format [,...]` | Formats to offer first, in priority order, e.g. `flac:48000:24:2,pcm:48000:24:2`. Everything else the player normally offers still follows, and a server uses the first format it can encode, so it can still fall back to a later one. This sets a preference, not a restriction. | -| `--input ` | A capture device to stream to the server: a microphone or line-in, e.g. `default` or `hw:1,0`. `-l` lists them. Off unless given. See [Stream a microphone or line-in](https://github.com/Sendspin/sendspin-cpp-cli/wiki/Advanced-Usage#stream-a-microphone-or-line-in). | +| `--input ` | A capture device to stream to the server: a microphone or line-in, e.g. `default`, `hw:1,0`, `pulse` or `pipewire:`. `-l` lists them. Off unless given. See [Stream a microphone or line-in](https://github.com/Sendspin/sendspin-cpp-cli/wiki/Advanced-Usage#stream-a-microphone-or-line-in). | | `--allow-unpaired` | Let a server play without pairing with this player first. Off by default. | | `--pairing-code <8 digits>` | A fixed code a server can pair with, confirmed with `sendspin-cli pair confirm`. Without it the player shows a fresh 6-digit code per attempt. | | `-s, --server ` | Connect out to a server rather than waiting to be discovered, or `mdns:[]` to discover one over mDNS. Turns off the mDNS advertisement. | diff --git a/docs/wiki/Advanced-Usage.md b/docs/wiki/Advanced-Usage.md index 91f8d8c..ed92442 100644 --- a/docs/wiki/Advanced-Usage.md +++ b/docs/wiki/Advanced-Usage.md @@ -60,33 +60,47 @@ streams that audio to the server, for a turntable on a line-in or a microphone i ```bash sendspin-cli -n den --input default sendspin-cli -n den --input hw:1,0 +sendspin-cli -n den --input pipewire +sendspin-cli -n den --input pulse:alsa_input.usb-mic.analog-stereo ``` Nothing is captured until the server starts the source, and the device is released again when the server stops it. Without `--input` the player does not offer the source role at all. -`sendspin-cli -l` lists capture devices in their own section, under the outputs, with the -rates, formats and channel counts each takes. `--input` reads its argument the way `-o` -does: `alsa:`, or a bare ALSA name such as `default`, `hw:1,0` or `plughw:1,0`. -Two names need no sound card: `tone` streams a 440 Hz test tone, which checks the path to -the server without a microphone, and `null` streams silence. +`sendspin-cli -l` lists capture devices in their own section, under the outputs: ALSA +PCMs with the rates, formats and channel counts each takes, then the PulseAudio sources +and PipeWire source nodes. `--input` reads its argument the way `-o` does: + +- `alsa:`, or a bare ALSA name such as `default`, `hw:1,0` or `plughw:1,0`. +- `pulse:` for a PulseAudio source, or `pulse` alone for the server's default. +- `pipewire:` for a PipeWire source node, or `pipewire` alone for the graph's + default. `pulse:default` and `pipewire:default` mean the same as the bare names. +- `tone` and `null`, which need no sound card: `tone` streams a 440 Hz test tone, which + checks the path to the server without a microphone, and `null` streams silence. + +On a desktop, prefer `pulse` or `pipewire` to a `hw:` name: the sound server shares the +card with other programs, where a `hw:` device is held exclusively. The capture format is chosen once at startup: 48000 Hz, stereo, 16-bit if the device -takes it, otherwise the nearest it does take — a mono microphone is sent as mono. The -startup log and `sendspin-cli status` both show what was chosen: +takes it, otherwise the nearest it does take — a mono microphone is sent as mono. A sound +server converts, so a capture through `pulse` or `pipewire` always runs at 48000 Hz, +stereo, 16-bit. The startup log and `sendspin-cli status` both show what was chosen: ```text input: hw:1,0 (48000 Hz / 2 ch / 16-bit), streaming ``` -A device that cannot be opened — a wrong name, or a card another program holds — stops -the player at startup with an error naming it. A device unplugged mid-stream is logged -and reopened when it comes back; playback is not affected. +A device that cannot be opened — a wrong name, a card another program holds, or a sound +server that is not running — stops the player at startup with an error naming it. A +device unplugged mid-stream, or a sound server that restarts, is logged and reopened when +it comes back; playback is not affected. A named PulseAudio source or PipeWire node is +never swapped for the default while it is away. -Capture through PulseAudio, PipeWire, CoreAudio or PortAudio is not available yet, so -capturing from a sound card needs a build with ALSA; `tone` and `null` work in any build. -ALSA's own `pulse` and `pipewire` PCMs still work: `--input alsa:pulse`. +Capture through CoreAudio or PortAudio is not available yet, so capturing from a sound +card needs a build with ALSA, PulseAudio or PipeWire; `tone` and `null` work in any +build. A build without the PulseAudio or PipeWire backend can still reach that server +through ALSA's own PCMs: `--input alsa:pulse`, `--input alsa:pipewire`. ## Logging and background operation diff --git a/docs/wiki/Configuration.md b/docs/wiki/Configuration.md index 8bdd261..d5dede7 100644 --- a/docs/wiki/Configuration.md +++ b/docs/wiki/Configuration.md @@ -53,7 +53,7 @@ already does that. | `product-name` | `--product-name` | the product name `client/hello` reports | `sendspin-cli` | | `server` | `-s`, `--server` | `[:]`, a `ws://` URL, or `mdns:[]` | none — wait to be discovered | | `port` | `--port` | the port this player's own WebSocket server listens on | `8928` | -| `input` | `--input` | a capture device to stream to the server: `default`, `hw:1,0`, `alsa:`, `tone`, `null`; see [Stream a microphone or line-in](Advanced-Usage#stream-a-microphone-or-line-in) | none — nothing is captured | +| `input` | `--input` | a capture device to stream to the server: `default`, `hw:1,0`, `alsa:`, `pulse[:]`, `pipewire[:]`, `tone`, `null`; see [Stream a microphone or line-in](Advanced-Usage#stream-a-microphone-or-line-in) | none — nothing is captured | | `buffer-ms` | `--buffer-ms` | audio the output backend keeps queued, 10–2000 | `100` | | `audio-format` | `--audio-format` | preferred formats, comma-separated in priority order: `codec:rate:depth:channels[,...]`, e.g. `flac:48000:24:2,pcm:48000:24:2`; offered first in that order, with the rest of the advertised list still behind them, so a server that cannot encode them falls back — a preference, not a restriction; refuses to start if the advertised list does not carry every one — it carries a single channel count — and an `opus` entry at anything but 48000/16 and at most 2 channels is refused outright | none — device-derived order | | `output-delay` | `--output-delay` | latency this endpoint's hardware adds after the audio port, 0–5000. `static-delay` / `--static-delay` is still accepted as the old name | `0` | diff --git a/packaging/sendspin-cli.conf.example b/packaging/sendspin-cli.conf.example index 6c93e99..00c02ff 100644 --- a/packaging/sendspin-cli.conf.example +++ b/packaging/sendspin-cli.conf.example @@ -18,6 +18,7 @@ #output = default # Capture device to stream to the server (a microphone or line-in); unset captures nothing. +# pulse or pipewire captures the sound server's default source; -l lists the named ones. #input = hw:1,0 # Audio the output backend keeps buffered, 10-2000 ms. Raise it on a device that stutters. From b66c4c24df68e216d213484cd7896ae69dd07a2e Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 7 Oct 2026 11:08:34 -0500 Subject: [PATCH 4/6] Bound the source name check on reopen and take duplex nodes A reopen ran the name check on the 3 s startup timeout, so a hung server could hold up a stop. A refused PipeWire stream now ends the open wait at once, and Audio/Duplex nodes can be named. --- src/pipewire_sink.cpp | 9 ++++-- src/pipewire_sink.h | 4 ++- src/pipewire_source.cpp | 68 +++++++++++++++++++++-------------------- src/pipewire_source.h | 2 +- src/pulse_sink.cpp | 4 +-- src/pulse_sink.h | 3 +- src/pulse_source.cpp | 27 ++++++++-------- src/pulse_source.h | 2 +- 8 files changed, 63 insertions(+), 56 deletions(-) diff --git a/src/pipewire_sink.cpp b/src/pipewire_sink.cpp index ccb5a06..0a7ff3e 100644 --- a/src/pipewire_sink.cpp +++ b/src/pipewire_sink.cpp @@ -144,7 +144,7 @@ constexpr pw_core_events CORE_EVENTS = [] { } // namespace bool pipewire_list_nodes(std::initializer_list media_classes, - std::vector& out, std::string& error) { + std::vector& out, std::string& error, int timeout_ms) { pw_thread_loop* loop = pw_thread_loop_new("sendspin-pw-scan", nullptr); if (loop == nullptr) { error = "cannot create a PipeWire loop"; @@ -182,8 +182,11 @@ bool pipewire_list_nodes(std::initializer_list media_classes, pw_registry_add_listener(registry, ®istry_hook, ®ISTRY_EVENTS, &walk); // A round trip, so the wait ends when the daemon has finished reporting. walk.sync_seq = pw_core_sync(core, PW_ID_CORE, 0); + timespec deadline{}; + pw_thread_loop_get_time(loop, &deadline, + static_cast(timeout_ms) * 1000 * 1000); while (!walk.done) { - if (pw_thread_loop_timed_wait(loop, PIPEWIRE_TIMEOUT_S) != 0) { + if (pw_thread_loop_timed_wait_full(loop, &deadline) != 0) { break; } } @@ -205,7 +208,7 @@ bool pipewire_list_nodes(std::initializer_list media_classes, error = "the PipeWire daemon refused the registry walk: " + refusal; } else if (!answered) { error = "the PipeWire daemon did not finish listing its nodes within " + - std::to_string(PIPEWIRE_TIMEOUT_S) + " s"; + std::to_string(timeout_ms) + " ms"; } { diff --git a/src/pipewire_sink.h b/src/pipewire_sink.h index b76720d..1cbb168 100644 --- a/src/pipewire_sink.h +++ b/src/pipewire_sink.h @@ -102,8 +102,10 @@ struct PipeWireNode { /// Walks the daemon's registry once on a private loop for nodes of any of `media_classes`. /// Needs a live PipeWireGuard. +/// @param timeout_ms PIPEWIRE_RECOVERY_TIMEOUT_MS on paths a stop must not wait behind. bool pipewire_list_nodes(std::initializer_list media_classes, - std::vector& out, std::string& error); + std::vector& out, std::string& error, + int timeout_ms = PIPEWIRE_TIMEOUT_S * 1000); /// An AudioSink that plays through a PipeWire graph; `-o pipewire:` a node name or empty. /// process() reads the ring and format fields unlocked: change them only while disconnected. diff --git a/src/pipewire_source.cpp b/src/pipewire_source.cpp index abb53f5..341c4fb 100644 --- a/src/pipewire_source.cpp +++ b/src/pipewire_source.cpp @@ -61,9 +61,11 @@ spa_audio_format spa_format_for(uint8_t bit_depth) { } } -/// Lists the graph's capture nodes, virtual ones included. -bool list_source_nodes(std::vector& out, std::string& error) { - return pipewire_list_nodes({"Audio/Source", "Audio/Source/Virtual"}, out, error); +/// Lists the nodes a capture stream can link to. +bool list_source_nodes(std::vector& out, std::string& error, + int timeout_ms = PIPEWIRE_TIMEOUT_S * 1000) { + return pipewire_list_nodes({"Audio/Source", "Audio/Source/Virtual", "Audio/Duplex"}, out, error, + timeout_ms); } } // namespace @@ -99,15 +101,15 @@ void PipeWireAudioSource::list_devices(std::FILE* out) { } } -bool PipeWireAudioSource::find_node_(std::string& error) const { +bool PipeWireAudioSource::find_node_(std::string& error, int timeout_ms) const { std::vector nodes; - if (!list_source_nodes(nodes, error)) { + if (!list_source_nodes(nodes, error, timeout_ms)) { return false; } - const bool found = - this->device_.empty() || - std::any_of(nodes.begin(), nodes.end(), - [this](const PipeWireNode& node) { return node.name == this->device_; }); + const bool found = this->device_.empty() || + std::any_of(nodes.begin(), nodes.end(), [this](const PipeWireNode& node) { + return node.name == this->device_; + }); if (!found) { error = "--input " + this->name() + ": this graph has no audio source node by that name"; } @@ -115,7 +117,7 @@ bool PipeWireAudioSource::find_node_(std::string& error) const { } bool PipeWireAudioSource::negotiate(StreamFormat& format, std::string& error) { - if (!this->find_node_(error)) { + if (!this->find_node_(error, PIPEWIRE_TIMEOUT_S * 1000)) { error += " -- run with -l to list this host's capture devices"; return false; } @@ -134,7 +136,7 @@ bool PipeWireAudioSource::open(const StreamFormat& format) { } // A named node that has gone would otherwise be swapped for the default by the session manager. std::string error; - if (!this->device_.empty() && !this->find_node_(error)) { + if (!this->device_.empty() && !this->find_node_(error, PIPEWIRE_RECOVERY_TIMEOUT_MS)) { cli_log(LogLevel::ERROR, "pipewire: %s", error.c_str()); return false; } @@ -151,13 +153,12 @@ bool PipeWireAudioSource::open(const StreamFormat& format) { this->bytes_per_frame_ = static_cast(format.channels) * (static_cast(format.bit_depth) / 8U); // The spare byte the ring keeps to tell full from empty. - this->ring_.reset((static_cast(format.sample_rate) * RING_MS / 1000U * - this->bytes_per_frame_) + - 1); + this->ring_.reset( + (static_cast(format.sample_rate) * RING_MS / 1000U * this->bytes_per_frame_) + 1); - pw_properties* props = pw_properties_new( - PW_KEY_MEDIA_TYPE, "Audio", PW_KEY_MEDIA_CATEGORY, "Capture", PW_KEY_APP_NAME, - PIPEWIRE_APP_NAME, PW_KEY_NODE_NAME, PIPEWIRE_NODE_NAME, nullptr); + pw_properties* props = pw_properties_new(PW_KEY_MEDIA_TYPE, "Audio", PW_KEY_MEDIA_CATEGORY, + "Capture", PW_KEY_APP_NAME, PIPEWIRE_APP_NAME, + PW_KEY_NODE_NAME, PIPEWIRE_NODE_NAME, nullptr); if (props == nullptr) { cli_log(LogLevel::ERROR, "pipewire: cannot describe the input stream to the graph"); this->close(); @@ -203,19 +204,21 @@ bool PipeWireAudioSource::open(const StreamFormat& format) { this->stream_ = pw_stream_new_simple(pw_thread_loop_get_loop(this->loop_), PIPEWIRE_APP_NAME, props, &events, this); if (this->stream_ != nullptr) { - const auto flags = static_cast( - PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS); + const auto flags = static_cast(PW_STREAM_FLAG_AUTOCONNECT | + PW_STREAM_FLAG_MAP_BUFFERS | + PW_STREAM_FLAG_RT_PROCESS); err = pw_stream_connect(this->stream_, PW_DIRECTION_INPUT, PW_ID_ANY, flags, params, 1); } if (err >= 0) { // PAUSED counts: the graph has taken the stream and starts it when a source is linked. timespec deadline{}; - pw_thread_loop_get_time(this->loop_, &deadline, - static_cast(PIPEWIRE_RECOVERY_TIMEOUT_MS) * 1000 * 1000); + pw_thread_loop_get_time( + this->loop_, &deadline, + static_cast(PIPEWIRE_RECOVERY_TIMEOUT_MS) * 1000 * 1000); while (true) { const auto state = static_cast(this->stream_state_.load()); if (state == PW_STREAM_STATE_STREAMING || state == PW_STREAM_STATE_PAUSED || - state == PW_STREAM_STATE_ERROR) { + this->lost_.load()) { break; } if (pw_thread_loop_timed_wait_full(this->loop_, &deadline) != 0) { @@ -279,13 +282,13 @@ void PipeWireAudioSource::stream_process_cb(void* userdata) { pw_time time{}; if (pw_stream_get_time_n(self->stream_, &time, sizeof(time)) == 0 && time.rate.denom != 0) { - // `delay` is the newest frame's age; the first is older by this buffer and by - // what the converter still holds. - const double age_s = (static_cast(time.delay) * - static_cast(time.rate.num) / - static_cast(time.rate.denom)) + - (static_cast(time.buffered + frames) / - static_cast(self->rate_)); + // `delay` is the newest frame's age; the first is this buffer and the converter + // older. + const double age_s = + (static_cast(time.delay) * static_cast(time.rate.num) / + static_cast(time.rate.denom)) + + (static_cast(time.buffered + frames) / + static_cast(self->rate_)); // `now` is CLOCK_MONOTONIC, which is what steady_clock reads here. const int64_t first_us = (time.now / 1000) - static_cast(age_s * 1e6); self->origin_us_.store(first_us - static_cast(self->frames_written_) * @@ -324,10 +327,9 @@ int PipeWireAudioSource::read(uint8_t* data, size_t length, uint32_t timeout_ms, this->ring_.read(data, bytes); const int64_t origin_us = this->origin_us_.load(); - capture_time_us = - (origin_us == 0) - ? 0 - : origin_us + static_cast(this->frames_read_) * US_PER_S / this->rate_; + capture_time_us = (origin_us == 0) ? 0 + : origin_us + static_cast(this->frames_read_) * + US_PER_S / this->rate_; this->frames_read_ += bytes / this->bytes_per_frame_; return static_cast(bytes); } diff --git a/src/pipewire_source.h b/src/pipewire_source.h index e5c81c3..b55a62d 100644 --- a/src/pipewire_source.h +++ b/src/pipewire_source.h @@ -55,7 +55,7 @@ class PipeWireAudioSource final : public AudioSource { static void stream_process_cb(void* userdata); /// Checks that a daemon answers and has the named node. - bool find_node_(std::string& error) const; + bool find_node_(std::string& error, int timeout_ms) const; /// Declared first so it is destroyed last, after the stream and its loop. PipeWireGuard pw_; diff --git a/src/pulse_sink.cpp b/src/pulse_sink.cpp index 8949d21..6cf6235 100644 --- a/src/pulse_sink.cpp +++ b/src/pulse_sink.cpp @@ -137,12 +137,12 @@ std::string default_sink_name(PulseConnection& conn) { } // namespace -bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op) { +bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op, int timeout_ms) { if (op == nullptr) { return false; } const bool answered = - conn.wait_for([&query] { return query.done.load(std::memory_order_acquire); }); + conn.wait_for([&query] { return query.done.load(std::memory_order_acquire); }, timeout_ms); { const MainloopLock lock(conn.mainloop()); if (!answered) { diff --git a/src/pulse_sink.h b/src/pulse_sink.h index a9cc118..f129172 100644 --- a/src/pulse_sink.h +++ b/src/pulse_sink.h @@ -130,7 +130,8 @@ struct PulseQuery { /// Waits for one query, cancelling it on timeout so a late callback cannot touch a dead `query`. /// The mainloop lock must NOT be held. /// @return true if the query completed and the server answered it. -bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op); +bool await_query(PulseConnection& conn, PulseQuery& query, pa_operation* op, + int timeout_ms = PULSE_TIMEOUT_MS); /// An AudioSink that plays through a PulseAudio server; `-o pulse:` a sink name or empty. /// Lock order is mutex_ then the mainloop lock; no libpulse callback takes mutex_. diff --git a/src/pulse_source.cpp b/src/pulse_source.cpp index 6b629c6..d4b2998 100644 --- a/src/pulse_source.cpp +++ b/src/pulse_source.cpp @@ -95,7 +95,8 @@ void server_info_cb(pa_context* /*context*/, const pa_server_info* info, void* u } /// Asks the server for every source it has, monitors included. The mainloop lock must NOT be held. -bool list_sources(PulseConnection& conn, std::vector& out) { +bool list_sources(PulseConnection& conn, std::vector& out, + int timeout_ms = PULSE_TIMEOUT_MS) { SourceListQuery query; query.conn = &conn; pa_operation* op = nullptr; @@ -103,15 +104,14 @@ bool list_sources(PulseConnection& conn, std::vector& out) { const MainloopLock lock(conn.mainloop()); op = pa_context_get_source_info_list(conn.context(), source_info_cb, &query); } - if (!await_query(conn, query, op)) { + if (!await_query(conn, query, op, timeout_ms)) { return false; } out = std::move(query.sources); return true; } -/// Asks the server which source it calls default. Empty when it will not say. Lock must NOT be -/// held. +/// The source the server calls default, or empty. The mainloop lock must NOT be held. std::string default_source_name(PulseConnection& conn) { DefaultSourceQuery query; query.conn = &conn; @@ -164,9 +164,9 @@ void PulseAudioSource::list_devices(std::FILE* out) { } } -bool PulseAudioSource::find_source_(std::string& error) { +bool PulseAudioSource::find_source_(std::string& error, int timeout_ms) { std::vector sources; - if (!list_sources(this->conn_, sources)) { + if (!list_sources(this->conn_, sources, timeout_ms)) { error = "cannot list the sources on " + this->conn_.server_name(); return false; } @@ -181,7 +181,8 @@ bool PulseAudioSource::find_source_(std::string& error) { } bool PulseAudioSource::negotiate(StreamFormat& format, std::string& error) { - if (!this->conn_.connect(error) || (!this->device_.empty() && !this->find_source_(error))) { + if (!this->conn_.connect(error) || + (!this->device_.empty() && !this->find_source_(error, PULSE_TIMEOUT_MS))) { error += " -- run with -l to list this host's capture devices"; return false; } @@ -210,7 +211,7 @@ bool PulseAudioSource::open(const StreamFormat& format) { } // pipewire-pulse records from the default when the named source is missing. std::string error; - if (!this->device_.empty() && !this->find_source_(error)) { + if (!this->device_.empty() && !this->find_source_(error, PULSE_RECOVERY_TIMEOUT_MS)) { cli_log(LogLevel::ERROR, "pulse: %s", error.c_str()); return false; } @@ -345,12 +346,10 @@ int PulseAudioSource::read(uint8_t* data, size_t length, uint32_t timeout_ms, // Cleared before the server is asked, so a fragment landing in between still ends the wait. this->readable_.store(false); int bytes = this->drain_(data, length, capture_time_us); - if (bytes == 0 && this->conn_.wait_for( - [this] { - return this->readable_.load() || this->lost_.load() || - !this->conn_.ready(); - }, - static_cast(timeout_ms))) { + if (bytes == 0 && + this->conn_.wait_for( + [this] { return this->readable_.load() || this->lost_.load() || !this->conn_.ready(); }, + static_cast(timeout_ms))) { bytes = this->drain_(data, length, capture_time_us); } return bytes; diff --git a/src/pulse_source.h b/src/pulse_source.h index f76aae7..4f58cbe 100644 --- a/src/pulse_source.h +++ b/src/pulse_source.h @@ -50,7 +50,7 @@ class PulseAudioSource final : public AudioSource { static void stream_read_cb(pa_stream* stream, size_t nbytes, void* userdata); /// Checks that the connected server has the named source. - bool find_source_(std::string& error); + bool find_source_(std::string& error, int timeout_ms); /// Copies out what the server has queued, without waiting. /// @return Bytes copied, 0 if nothing is queued, negative once the stream is lost. From 27ef9abb21950d11e0a75215af895133a0c762de Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 7 Oct 2026 11:08:47 -0500 Subject: [PATCH 5/6] Keep the capture delay comment to one line --- src/pipewire_source.cpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/pipewire_source.cpp b/src/pipewire_source.cpp index 341c4fb..2fffcf5 100644 --- a/src/pipewire_source.cpp +++ b/src/pipewire_source.cpp @@ -282,8 +282,7 @@ void PipeWireAudioSource::stream_process_cb(void* userdata) { pw_time time{}; if (pw_stream_get_time_n(self->stream_, &time, sizeof(time)) == 0 && time.rate.denom != 0) { - // `delay` is the newest frame's age; the first is this buffer and the converter - // older. + // `delay` is the newest frame's age; add this buffer and the converter's hold. const double age_s = (static_cast(time.delay) * static_cast(time.rate.num) / static_cast(time.rate.denom)) + From d83901e22b667e7006cd26f5f89d4c32ec035a2c Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 7 Oct 2026 13:13:42 -0500 Subject: [PATCH 6/6] Open a capture stream at startup and keep timestamps exact after an overrun negotiate() now opens the real PipeWire or Pulse stream and closes it again, so a server that lists the source but will not record from it stops startup instead of retrying forever once a stream begins. A PipeWire quantum that does not fit the ring is counted, and read() drops what was queued before it, so later frames keep their true capture time. --- src/pipewire_source.cpp | 85 ++++++++++++++++++++++++++++------------- src/pipewire_source.h | 7 +++- src/pulse_source.cpp | 58 +++++++++++++++------------- src/pulse_source.h | 5 ++- 4 files changed, 101 insertions(+), 54 deletions(-) diff --git a/src/pipewire_source.cpp b/src/pipewire_source.cpp index 2fffcf5..8cb590e 100644 --- a/src/pipewire_source.cpp +++ b/src/pipewire_source.cpp @@ -42,7 +42,7 @@ constexpr const char* PIPEWIRE_NODE_NAME = "sendspin-cli-input"; /// Quantum asked of the graph; one source-role chunk. constexpr uint32_t QUANTUM_MS = 20; -/// Ring size; whole quanta are dropped once a stalled reader has let it fill. +/// Ring size; a reader stalled for longer loses what is queued and resumes at live audio. constexpr uint32_t RING_MS = 1000; constexpr int64_t US_PER_S = 1000000; @@ -117,34 +117,46 @@ bool PipeWireAudioSource::find_node_(std::string& error, int timeout_ms) const { } bool PipeWireAudioSource::negotiate(StreamFormat& format, std::string& error) { - if (!this->find_node_(error, PIPEWIRE_TIMEOUT_S * 1000)) { + settle_server_capture_format(format); + if (!this->open_(format, PIPEWIRE_TIMEOUT_S * 1000, error)) { error += " -- run with -l to list this host's capture devices"; return false; } - settle_server_capture_format(format); + this->close(); return true; } bool PipeWireAudioSource::open(const StreamFormat& format) { + std::string error; + if (!this->open_(format, PIPEWIRE_RECOVERY_TIMEOUT_MS, error)) { + cli_log(LogLevel::ERROR, "pipewire: %s", error.c_str()); + return false; + } + cli_log(LogLevel::INFO, "pipewire: input '%s' open at %u Hz, %u ch, %u-bit", + this->name().c_str(), format.sample_rate, format.channels, format.bit_depth); + return true; +} + +bool PipeWireAudioSource::open_(const StreamFormat& format, int timeout_ms, std::string& error) { this->close(); + const std::string device = "PipeWire input '" + this->name() + "'"; + const std::string wanted = std::to_string(format.sample_rate) + " Hz / " + + std::to_string(static_cast(format.channels)) + " ch / " + + std::to_string(static_cast(format.bit_depth)) + "-bit"; const spa_audio_format spa_format = spa_format_for(format.bit_depth); if (spa_format == SPA_AUDIO_FORMAT_UNKNOWN || format.channels == 0 || format.sample_rate == 0) { - cli_log(LogLevel::ERROR, "pipewire: input cannot capture %u Hz / %u ch / %u-bit", - format.sample_rate, format.channels, format.bit_depth); + error = device + " cannot capture " + wanted; return false; } // A named node that has gone would otherwise be swapped for the default by the session manager. - std::string error; - if (!this->device_.empty() && !this->find_node_(error, PIPEWIRE_RECOVERY_TIMEOUT_MS)) { - cli_log(LogLevel::ERROR, "pipewire: %s", error.c_str()); + if (!this->device_.empty() && !this->find_node_(error, timeout_ms)) { return false; } this->loop_ = pw_thread_loop_new("sendspin-pw-in", nullptr); if (this->loop_ == nullptr || pw_thread_loop_start(this->loop_) < 0) { - cli_log(LogLevel::ERROR, "pipewire: cannot start a loop for input '%s'", - this->name().c_str()); + error = "cannot start a loop for " + device; this->close(); return false; } @@ -160,7 +172,7 @@ bool PipeWireAudioSource::open(const StreamFormat& format) { "Capture", PW_KEY_APP_NAME, PIPEWIRE_APP_NAME, PW_KEY_NODE_NAME, PIPEWIRE_NODE_NAME, nullptr); if (props == nullptr) { - cli_log(LogLevel::ERROR, "pipewire: cannot describe the input stream to the graph"); + error = "cannot describe " + device + " to the graph"; this->close(); return false; } @@ -212,9 +224,8 @@ bool PipeWireAudioSource::open(const StreamFormat& format) { if (err >= 0) { // PAUSED counts: the graph has taken the stream and starts it when a source is linked. timespec deadline{}; - pw_thread_loop_get_time( - this->loop_, &deadline, - static_cast(PIPEWIRE_RECOVERY_TIMEOUT_MS) * 1000 * 1000); + pw_thread_loop_get_time(this->loop_, &deadline, + static_cast(timeout_ms) * 1000 * 1000); while (true) { const auto state = static_cast(this->stream_state_.load()); if (state == PW_STREAM_STATE_STREAMING || state == PW_STREAM_STATE_PAUSED || @@ -230,22 +241,19 @@ bool PipeWireAudioSource::open(const StreamFormat& format) { const auto state = static_cast(this->stream_state_.load()); if (err < 0) { - cli_log(LogLevel::ERROR, "pipewire: cannot capture from '%s': %s", this->name().c_str(), - spa_strerror(err)); - } else if (state == PW_STREAM_STATE_ERROR) { - err = -EIO; // stream_state_cb() has said why + error = "cannot open " + device + ": " + spa_strerror(err); + } else if (this->lost_.load()) { + err = -EIO; + error = "the graph refused " + device + " at " + wanted; } else if (state != PW_STREAM_STATE_STREAMING && state != PW_STREAM_STATE_PAUSED) { err = -ETIMEDOUT; - cli_log(LogLevel::ERROR, "pipewire: input '%s' would not start at %u Hz, %u ch, %u-bit", - this->name().c_str(), format.sample_rate, format.channels, format.bit_depth); + error = device + " would not start at " + wanted + " within " + std::to_string(timeout_ms) + + " ms"; } if (err < 0) { this->close(); return false; } - - cli_log(LogLevel::INFO, "pipewire: input '%s' open at %u Hz, %u ch, %u-bit", - this->name().c_str(), format.sample_rate, format.channels, format.bit_depth); return true; } @@ -277,8 +285,7 @@ void PipeWireAudioSource::stream_process_cb(void* userdata) { if (data.data != nullptr && stride != 0) { const uint32_t offset = std::min(data.chunk->offset, data.maxsize); const size_t frames = std::min(data.chunk->size, data.maxsize - offset) / stride; - // All or nothing, so the frames in the ring stay contiguous within a quantum. - if (frames != 0 && self->ring_.free_space() >= frames * stride) { + if (frames != 0) { pw_time time{}; if (pw_stream_get_time_n(self->stream_, &time, sizeof(time)) == 0 && time.rate.denom != 0) { @@ -293,7 +300,12 @@ void PipeWireAudioSource::stream_process_cb(void* userdata) { self->origin_us_.store(first_us - static_cast(self->frames_written_) * US_PER_S / self->rate_); } - self->ring_.write(static_cast(data.data) + offset, frames * stride); + // All or nothing: a quantum that does not fit is counted, and read() skips to it. + if (self->ring_.free_space() >= frames * stride) { + self->ring_.write(static_cast(data.data) + offset, frames * stride); + } else { + self->frames_dropped_.fetch_add(frames); + } self->frames_written_ += frames; } } @@ -318,6 +330,26 @@ int PipeWireAudioSource::read(uint8_t* data, size_t length, uint32_t timeout_ms, return -1; } + const uint64_t dropped = this->frames_dropped_.exchange(0); + if (dropped != 0) { + // Everything queued predates the drop, so it goes too and the frame count stays exact. + size_t stale = this->ring_.available(); + stale -= stale % this->bytes_per_frame_; + const uint64_t skipped = dropped + (stale / this->bytes_per_frame_); + while (stale > 0) { + const size_t got = this->ring_.read(data, std::min(stale, length)); + if (got == 0) { + break; + } + stale -= got; + } + this->frames_read_ += skipped; + cli_log(LogLevel::WARN, "pipewire: input '%s' overran -- skipped %llu ms of audio", + this->name().c_str(), + static_cast(skipped * 1000U / this->rate_)); + return 0; + } + size_t bytes = std::min(this->ring_.available(), length); bytes -= bytes % this->bytes_per_frame_; if (bytes == 0) { @@ -352,6 +384,7 @@ void PipeWireAudioSource::close() { this->bytes_per_frame_ = 0; this->frames_written_ = 0; this->frames_read_ = 0; + this->frames_dropped_.store(0); this->origin_us_.store(0); this->lost_.store(false); this->stream_state_.store(PW_STREAM_STATE_UNCONNECTED); diff --git a/src/pipewire_source.h b/src/pipewire_source.h index b55a62d..1df9f82 100644 --- a/src/pipewire_source.h +++ b/src/pipewire_source.h @@ -32,7 +32,7 @@ namespace sendspin_cli { -/// An AudioSource that captures from a PipeWire graph; `--input pipewire:` a node name or empty. +/// An AudioSource that captures from a PipeWire graph; `--input pipewire:` takes a node name. /// process() writes the ring and reads the format fields unlocked: change them only while closed. class PipeWireAudioSource final : public AudioSource { public: @@ -54,6 +54,9 @@ class PipeWireAudioSource final : public AudioSource { enum pw_stream_state state, const char* error); static void stream_process_cb(void* userdata); + /// Connects the stream, waiting up to `timeout_ms` for the graph to take it. + bool open_(const StreamFormat& format, int timeout_ms, std::string& error); + /// Checks that a daemon answers and has the named node. bool find_node_(std::string& error, int timeout_ms) const; @@ -70,6 +73,8 @@ class PipeWireAudioSource final : public AudioSource { uint64_t frames_written_{0}; ///< process() only uint64_t frames_read_{0}; ///< read() only + /// Frames process() could not fit in the ring, for read() to skip past. + std::atomic frames_dropped_{0}; /// Steady-clock time of this stream's frame 0 as process() last measured it; 0 until it has. std::atomic origin_us_{0}; diff --git a/src/pulse_source.cpp b/src/pulse_source.cpp index d4b2998..b603c3b 100644 --- a/src/pulse_source.cpp +++ b/src/pulse_source.cpp @@ -181,38 +181,47 @@ bool PulseAudioSource::find_source_(std::string& error, int timeout_ms) { } bool PulseAudioSource::negotiate(StreamFormat& format, std::string& error) { - if (!this->conn_.connect(error) || - (!this->device_.empty() && !this->find_source_(error, PULSE_TIMEOUT_MS))) { + settle_server_capture_format(format); + if (!this->open_(format, PULSE_TIMEOUT_MS, error)) { error += " -- run with -l to list this host's capture devices"; return false; } - settle_server_capture_format(format); + this->close(); return true; } bool PulseAudioSource::open(const StreamFormat& format) { + std::string error; + if (!this->open_(format, PULSE_RECOVERY_TIMEOUT_MS, error)) { + cli_log(LogLevel::ERROR, "pulse: %s", error.c_str()); + return false; + } + cli_log(LogLevel::INFO, "pulse: input '%s' open on %s at %u Hz, %u ch, %u-bit", + this->name().c_str(), this->conn_.server_name().c_str(), format.sample_rate, + format.channels, format.bit_depth); + return true; +} + +bool PulseAudioSource::open_(const StreamFormat& format, int timeout_ms, std::string& error) { this->close(); + const std::string device = "PulseAudio input '" + this->name() + "'"; + const std::string wanted = std::to_string(format.sample_rate) + " Hz / " + + std::to_string(static_cast(format.channels)) + " ch / " + + std::to_string(static_cast(format.bit_depth)) + "-bit"; pa_sample_spec spec{}; spec.format = pulse_format_for(format.bit_depth); spec.rate = format.sample_rate; spec.channels = format.channels; if (pa_sample_spec_valid(&spec) == 0) { - cli_log(LogLevel::ERROR, "pulse: input cannot capture %u Hz / %u ch / %u-bit", - format.sample_rate, format.channels, format.bit_depth); + error = device + " cannot capture " + wanted; return false; } - if (!this->conn_.ready()) { - std::string error; - if (!this->conn_.connect(error, PULSE_RECOVERY_TIMEOUT_MS)) { - cli_log(LogLevel::ERROR, "pulse: %s", error.c_str()); - return false; - } + if (!this->conn_.ready() && !this->conn_.connect(error, timeout_ms)) { + return false; } // pipewire-pulse records from the default when the named source is missing. - std::string error; - if (!this->device_.empty() && !this->find_source_(error, PULSE_RECOVERY_TIMEOUT_MS)) { - cli_log(LogLevel::ERROR, "pulse: %s", error.c_str()); + if (!this->device_.empty() && !this->find_source_(error, timeout_ms)) { return false; } @@ -232,7 +241,7 @@ bool PulseAudioSource::open(const StreamFormat& format) { // Unplugging the source ends the stream, so it is reopened rather than moved elsewhere. flags |= PA_STREAM_DONT_MOVE; } - const char* device = this->device_.empty() ? nullptr : this->device_.c_str(); + const char* source = this->device_.empty() ? nullptr : this->device_.c_str(); int err = -1; { @@ -241,12 +250,12 @@ bool PulseAudioSource::open(const StreamFormat& format) { if (this->stream_ != nullptr) { pa_stream_set_state_callback(this->stream_, &PulseAudioSource::stream_state_cb, this); pa_stream_set_read_callback(this->stream_, &PulseAudioSource::stream_read_cb, this); - err = pa_stream_connect_record(this->stream_, device, &attr, + err = pa_stream_connect_record(this->stream_, source, &attr, static_cast(flags)); } if (err < 0) { - cli_log(LogLevel::ERROR, "pulse: cannot capture from '%s': %s", this->name().c_str(), - pa_strerror(pa_context_errno(this->conn_.context()))); + error = "cannot open " + device + ": " + + pa_strerror(pa_context_errno(this->conn_.context())); } } if (err < 0) { @@ -259,18 +268,15 @@ bool PulseAudioSource::open(const StreamFormat& format) { const auto state = static_cast(this->stream_state_.load()); return state == PA_STREAM_READY || PA_STREAM_IS_GOOD(state) == 0; }, - PULSE_RECOVERY_TIMEOUT_MS); + timeout_ms); if (!settled || this->stream_state_.load() != PA_STREAM_READY) { - cli_log(LogLevel::ERROR, "pulse: input '%s' would not start at %u Hz, %u ch, %u-bit -- %s", - this->name().c_str(), format.sample_rate, format.channels, format.bit_depth, - settled ? "the server refused it" : "the server did not answer"); + error = + device + " would not start at " + wanted + ": " + + (settled ? "the server refused it" + : "the server did not answer within " + std::to_string(timeout_ms) + " ms"); this->close(); return false; } - - cli_log(LogLevel::INFO, "pulse: input '%s' open on %s at %u Hz, %u ch, %u-bit", - this->name().c_str(), this->conn_.server_name().c_str(), format.sample_rate, - format.channels, format.bit_depth); return true; } diff --git a/src/pulse_source.h b/src/pulse_source.h index 4f58cbe..0ad903c 100644 --- a/src/pulse_source.h +++ b/src/pulse_source.h @@ -29,7 +29,7 @@ namespace sendspin_cli { -/// An AudioSource that captures from a PulseAudio server; `--input pulse:` a source name or empty. +/// An AudioSource that captures from a PulseAudio server; `--input pulse:` takes a source name. class PulseAudioSource final : public AudioSource { public: /// @param device Source name, or empty for the server's default. @@ -49,6 +49,9 @@ class PulseAudioSource final : public AudioSource { static void stream_state_cb(pa_stream* stream, void* userdata); static void stream_read_cb(pa_stream* stream, size_t nbytes, void* userdata); + /// Connects if need be and opens the record stream, each wait bounded by `timeout_ms`. + bool open_(const StreamFormat& format, int timeout_ms, std::string& error); + /// Checks that the connected server has the named source. bool find_source_(std::string& error, int timeout_ms);