diff --git a/CMakeLists.txt b/CMakeLists.txt index ab49d6f..12fe93f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -318,13 +318,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/README.md b/README.md index 5e0a981..a1f375b 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`, or `coreaudio` on macOS. `-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:`, or `coreaudio` on macOS. `-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 0783830..0ff6910 100644 --- a/docs/wiki/Advanced-Usage.md +++ b/docs/wiki/Advanced-Usage.md @@ -60,6 +60,8 @@ 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 ``` On macOS: @@ -73,31 +75,44 @@ Nothing is captured until the server starts the source, and the device is releas 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`. -On macOS it is `coreaudio` for the system default input, which is followed if it moves, or -`coreaudio:` for one device from `-l`'s list. -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, or on macOS the CoreAudio input devices. `--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. +- On macOS, `coreaudio` for the system default input, which is followed if it moves, or + `coreaudio:` for one device from `-l`'s list. +- `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 Linux 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. On -macOS the rate is the one the device is currently set to, because CoreAudio does not -resample capture; change it in Audio MIDI Setup before starting. The +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. On macOS the rate is the one the device is currently set to, because +CoreAudio does not resample capture; change it in Audio MIDI Setup before starting. 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 or PortAudio is not available yet, so capturing from -a sound card needs ALSA on Linux or CoreAudio on macOS; `tone` and `null` work in any -build. ALSA's own `pulse` and `pipewire` PCMs still work: `--input alsa:pulse`. +Capture through PortAudio is not available yet, so capturing from a sound card needs +ALSA, PulseAudio or PipeWire on Linux, or CoreAudio on macOS; `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`. ### Microphone permission on macOS diff --git a/docs/wiki/Configuration.md b/docs/wiki/Configuration.md index 121403b..6b3ca1b 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:`, `coreaudio[:]` on macOS, `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[:]`, `coreaudio[:]` on macOS, `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. diff --git a/src/audio_source.cpp b/src/audio_source.cpp index dac8080..65c3b15 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 #ifdef SENDSPIN_CLI_HAVE_COREAUDIO #include "coreaudio_source.h" #endif @@ -48,7 +54,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[] = {"portaudio", "pulse", "pipewire"}; +constexpr const char* UNSUPPORTED_BACKENDS[] = {"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 @@ -163,12 +180,31 @@ std::string input_backend_list() { #ifdef SENDSPIN_CLI_HAVE_ALSA list += ", alsa"; #endif + if (HAVE_PULSE) { + list += ", pulse"; + } + if (HAVE_PIPEWIRE) { + list += ", pipewire"; + } #ifdef SENDSPIN_CLI_HAVE_COREAUDIO list += ", coreaudio"; #endif 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) { if (spec.empty()) { error = "empty input device -- run with -l to list what this build has"; @@ -207,6 +243,21 @@ bool resolve_input_spec(const std::string& spec, InputSpec& out, std::string& er #endif } + 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 + error += ". ALSA's plugin PCM of that name still works: --input alsa:" + prefix; +#endif + return false; + } + out = {pulse ? SourceBackend::Pulse : SourceBackend::PipeWire, + rest == "default" ? std::string() : rest}; + return true; + } + if (prefix == "coreaudio") { #ifdef SENDSPIN_CLI_HAVE_COREAUDIO if (has_device && rest.empty()) { @@ -226,17 +277,11 @@ bool resolve_input_spec(const std::string& spec, InputSpec& out, std::string& er } 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(); -#ifdef SENDSPIN_CLI_HAVE_ALSA - if (prefix == "pulse" || prefix == "pipewire") { - error += ". ALSA's plugin PCM of that name still works: --input alsa:" + prefix; + if (prefix == name) { + error = "capturing through the " + prefix + + " backend is not supported yet -- --input takes: " + input_backend_list(); + return false; } -#endif - return false; } #ifdef SENDSPIN_CLI_HAVE_ALSA @@ -277,6 +322,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; case SourceBackend::CoreAudio: @@ -320,6 +375,30 @@ void print_capture_devices(std::FILE* out) { "plug-style PCM -- default, plughw: -- converts, so it takes the preferred\n" "format whatever the hardware behind it captures.\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 + #ifdef SENDSPIN_CLI_HAVE_COREAUDIO std::fprintf(out, "\nCoreAudio input devices on this host (--input coreaudio:):\n"); CoreAudioSource::list_devices(out); @@ -329,9 +408,10 @@ void print_capture_devices(std::FILE* out) { "16-bit. Audio MIDI Setup changes the rate. macOS asks for microphone access\n" "the first time --input opens a device.\n"); #endif -#if !defined(SENDSPIN_CLI_HAVE_ALSA) && !defined(SENDSPIN_CLI_HAVE_COREAUDIO) - std::fprintf(out, "\nThis build has no ALSA backend, so there is no sound card to capture\n" - "from here.\n"); +#if !defined(SENDSPIN_CLI_HAVE_ALSA) && !defined(SENDSPIN_CLI_HAVE_PULSE) && \ + !defined(SENDSPIN_CLI_HAVE_PIPEWIRE) && !defined(SENDSPIN_CLI_HAVE_COREAUDIO) + std::fprintf(out, "\nThis build has no ALSA, PulseAudio, PipeWire or CoreAudio backend, so\n" + "there is no sound card to capture from here.\n"); #endif } diff --git a/src/audio_source.h b/src/audio_source.h index 2a545f4..947c407 100644 --- a/src/audio_source.h +++ b/src/audio_source.h @@ -116,13 +116,15 @@ enum class SourceBackend { 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 CoreAudio, ///< a CoreAudio device by index or name, or this host's default input if empty }; /// 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 and CoreAudio's default + std::string device; ///< empty for the device-less sources and a backend's default }; /// Resolves an --input spec without opening anything, by -o's rules: a bare backend name, @@ -130,12 +132,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_sink.cpp b/src/pipewire_sink.cpp index 1d33aee..0a7ff3e 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, int timeout_ms) { 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; @@ -197,8 +182,11 @@ bool walk_registry(std::vector& out, std::string& error) { 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; } } @@ -220,7 +208,7 @@ bool walk_registry(std::vector& out, std::string& error) { 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"; } { @@ -246,8 +234,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 +291,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 +315,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..1cbb168 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,36 @@ 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. +/// @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, + 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. /// Lock order is mutex_ then the thread-loop lock; no PipeWire callback takes mutex_. diff --git a/src/pipewire_source.cpp b/src/pipewire_source.cpp new file mode 100644 index 0000000..8cb590e --- /dev/null +++ b/src/pipewire_source.cpp @@ -0,0 +1,393 @@ +// 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; 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; + +/// 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 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 + +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, int timeout_ms) const { + std::vector nodes; + 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_; + }); + 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) { + 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; + } + 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) { + error = device + " cannot capture " + wanted; + return false; + } + // A named node that has gone would otherwise be swapped for the default by the session manager. + 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) { + error = "cannot start a loop for " + device; + 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) { + error = "cannot describe " + device + " 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(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 || + this->lost_.load()) { + 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) { + 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; + error = device + " would not start at " + wanted + " within " + std::to_string(timeout_ms) + + " ms"; + } + if (err < 0) { + this->close(); + return false; + } + 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; + if (frames != 0) { + 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; 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)) + + (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_); + } + // 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; + } + } + 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; + } + + 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) { + 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->frames_dropped_.store(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..1df9f82 --- /dev/null +++ b/src/pipewire_source.h @@ -0,0 +1,90 @@ +// 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:` 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: + /// @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); + + /// 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; + + /// 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 + /// 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}; + + /// 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_sink.cpp b/src/pulse_sink.cpp index d6b8871..6cf6235 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, int timeout_ms) { + if (op == nullptr) { + return false; + } + const bool answered = + conn.wait_for([&query] { return query.done.load(std::memory_order_acquire); }, timeout_ms); + { + 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..f129172 100644 --- a/src/pulse_sink.h +++ b/src/pulse_sink.h @@ -96,6 +96,43 @@ 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, + 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_. class PulseAudioSink final : public AudioSink { diff --git a/src/pulse_source.cpp b/src/pulse_source.cpp new file mode 100644 index 0000000..b603c3b --- /dev/null +++ b/src/pulse_source.cpp @@ -0,0 +1,382 @@ +// 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, + int timeout_ms = PULSE_TIMEOUT_MS) { + 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, timeout_ms)) { + return false; + } + out = std::move(query.sources); + return true; +} + +/// 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; + 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, int timeout_ms) { + std::vector sources; + if (!list_sources(this->conn_, sources, timeout_ms)) { + 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) { + 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; + } + 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) { + error = device + " cannot capture " + wanted; + 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. + if (!this->device_.empty() && !this->find_source_(error, timeout_ms)) { + 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* source = 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_, source, &attr, + static_cast(flags)); + } + if (err < 0) { + error = "cannot open " + device + ": " + + 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; + }, + timeout_ms); + if (!settled || this->stream_state_.load() != PA_STREAM_READY) { + 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; + } + 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..0ad903c --- /dev/null +++ b/src/pulse_source.h @@ -0,0 +1,79 @@ +// 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:` takes a source name. +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); + + /// 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); + + /// 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 ee3ec07..599e316 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,7 +147,7 @@ TEST(ResolveInputSpec, DeviceLessNamesRefuseADevice) { } TEST(ResolveInputSpec, OutputOnlyBackendsAreRefusedByName) { - for (const char* spec : {"pulse", "pulse:mic", "pipewire:node", "portaudio:1"}) { + for (const char* spec : {"portaudio", "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; @@ -179,6 +180,84 @@ TEST(ResolveInputSpec, CoreaudioSaysItIsNotInThisBuild) { } #endif +#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());