diff --git a/apps/desktop/src/gui/application/application_state.h b/apps/desktop/src/gui/application/application_state.h index 4d7f0541..dab41043 100644 --- a/apps/desktop/src/gui/application/application_state.h +++ b/apps/desktop/src/gui/application/application_state.h @@ -66,8 +66,6 @@ struct InteractionState { bool mouse_controller_is_started_ = false; bool start_screen_capturer_ = false; bool screen_capturer_is_started_ = false; - bool start_speaker_capturer_ = false; - bool speaker_capturer_is_started_ = false; bool start_keyboard_capturer_ = false; bool show_cursor_ = false; bool keyboard_capturer_is_started_ = false; @@ -198,4 +196,4 @@ struct ApplicationState : MainWindowState, } // namespace crossdesk::gui_detail -#endif \ No newline at end of file +#endif diff --git a/apps/desktop/src/gui/features/devices/session_device_manager.cpp b/apps/desktop/src/gui/features/devices/session_device_manager.cpp index 75915ea9..85b91429 100644 --- a/apps/desktop/src/gui/features/devices/session_device_manager.cpp +++ b/apps/desktop/src/gui/features/devices/session_device_manager.cpp @@ -9,6 +9,7 @@ #include "platform.h" #include "rd_log.h" #include "runtime/gui_runtime.h" +#include "speaker_capturer_factory.h" namespace crossdesk { namespace { @@ -19,7 +20,23 @@ constexpr auto kFrameDeadlineTolerance = std::chrono::milliseconds(1); } // namespace -SessionDeviceManager::SessionDeviceManager(GuiRuntime &owner) : owner_(owner) {} +SessionDeviceManager::SessionDeviceManager(GuiRuntime &owner) + : owner_(owner), + speaker_capture_( + [] { + return std::unique_ptr( + SpeakerCapturerFactory().Create()); + }, + [this](unsigned char *data, size_t size, const char *) { + if (!owner_.peer_) return; + MiniRtcAudioFrame frame{}; + frame.data = reinterpret_cast(data); + frame.size = size; + frame.captured_timestamp = GetSystemTimeMicros(owner_.peer_); + SendAudioFrame(owner_.peer_, &frame, owner_.audio_label_.c_str()); + }) {} + +SessionDeviceManager::~SessionDeviceManager() { speaker_capture_.Shutdown(); } bool SessionDeviceManager::ShouldSendCapturedFrame( std::chrono::steady_clock::time_point now, int fps) { @@ -41,7 +58,6 @@ bool SessionDeviceManager::ShouldSendCapturedFrame( void SessionDeviceManager::Initialize() { InitializeAudioOutput(); screen_capturer_factory_ = new ScreenCapturerFactory(); - speaker_capturer_factory_ = new SpeakerCapturerFactory(); device_controller_factory_ = new DeviceControllerFactory(); keyboard_capturer_ = static_cast(device_controller_factory_->Create( @@ -200,45 +216,12 @@ int SessionDeviceManager::StopScreenCapturer() { return 0; } -int SessionDeviceManager::StartSpeakerCapturer() { - if (!speaker_capturer_) { - speaker_capturer_ = - static_cast(speaker_capturer_factory_->Create()); - const int init_ret = speaker_capturer_->Init( - [this](unsigned char *data, size_t size, const char *audio_name) { - MiniRtcAudioFrame frame{}; - frame.data = reinterpret_cast(data); - frame.size = size; - frame.captured_timestamp = GetSystemTimeMicros(owner_.peer_); - SendAudioFrame(owner_.peer_, &frame, owner_.audio_label_.c_str()); - }); - - if (init_ret != 0) { - speaker_capturer_->Destroy(); - delete speaker_capturer_; - speaker_capturer_ = nullptr; - } - } - - if (!speaker_capturer_) { - return -1; - } - - const int ret = speaker_capturer_->Start(); - if (ret != 0) { - LOG_ERROR("Start speaker capturer failed: {}", ret); - return ret; - } - owner_.start_speaker_capturer_ = true; - return 0; +void SessionDeviceManager::StartSpeakerCapturer() { + speaker_capture_.SetEnabled(true); } -int SessionDeviceManager::StopSpeakerCapturer() { - if (speaker_capturer_) { - speaker_capturer_->Stop(); - owner_.start_speaker_capturer_ = false; - } - return 0; +void SessionDeviceManager::StopSpeakerCapturer() { + speaker_capture_.SetEnabled(false); } int SessionDeviceManager::StartMouseController() { @@ -448,16 +431,6 @@ void SessionDeviceManager::UpdateInteractions() { owner_.screen_capturer_is_started_ = false; } - if (owner_.start_speaker_capturer_ && !owner_.speaker_capturer_is_started_) { - if (StartSpeakerCapturer() == 0) { - owner_.speaker_capturer_is_started_ = true; - } - } else if (!owner_.start_speaker_capturer_ && - owner_.speaker_capturer_is_started_) { - StopSpeakerCapturer(); - owner_.speaker_capturer_is_started_ = false; - } - if (owner_.start_mouse_controller_ && !owner_.mouse_controller_is_started_) { if (StartMouseController() == 0) { owner_.mouse_controller_is_started_ = true; @@ -556,6 +529,7 @@ SessionDeviceManager::display_info_list() const { } void SessionDeviceManager::DestroyDevices() { + speaker_capture_.Shutdown(); if (mouse_controller_) { mouse_controller_->Destroy(); delete mouse_controller_; @@ -566,11 +540,6 @@ void SessionDeviceManager::DestroyDevices() { delete screen_capturer_; screen_capturer_ = nullptr; } - if (speaker_capturer_) { - speaker_capturer_->Destroy(); - delete speaker_capturer_; - speaker_capturer_ = nullptr; - } if (keyboard_capturer_) { delete keyboard_capturer_; keyboard_capturer_ = nullptr; @@ -580,8 +549,6 @@ void SessionDeviceManager::DestroyDevices() { void SessionDeviceManager::DestroyFactories() { delete screen_capturer_factory_; screen_capturer_factory_ = nullptr; - delete speaker_capturer_factory_; - speaker_capturer_factory_ = nullptr; delete device_controller_factory_; device_controller_factory_ = nullptr; } diff --git a/apps/desktop/src/gui/features/devices/session_device_manager.h b/apps/desktop/src/gui/features/devices/session_device_manager.h index 9b139df8..ddbfa37c 100644 --- a/apps/desktop/src/gui/features/devices/session_device_manager.h +++ b/apps/desktop/src/gui/features/devices/session_device_manager.h @@ -1,5 +1,12 @@ -#ifndef CROSSDESK_GUI_SESSION_DEVICE_MANAGER_H_ -#define CROSSDESK_GUI_SESSION_DEVICE_MANAGER_H_ +/* + * @Author: DI JUNKUN + * @Date: 2026-09-09 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _SESSION_DEVICE_MANAGER_H_ +#define _SESSION_DEVICE_MANAGER_H_ + #include @@ -17,7 +24,7 @@ #include "device_controller_factory.h" #include "display_info.h" #include "screen_capturer_factory.h" -#include "speaker_capturer_factory.h" +#include "speaker_capture_controller.h" namespace crossdesk { @@ -29,6 +36,7 @@ class GuiRuntime; class SessionDeviceManager { public: explicit SessionDeviceManager(GuiRuntime &owner); + ~SessionDeviceManager(); void Initialize(); void UpdateInteractions(); @@ -38,8 +46,10 @@ public: int InitializeScreenCapturer(); int StartScreenCapturer(); int StopScreenCapturer(); - int StartSpeakerCapturer(); - int StopSpeakerCapturer(); + // Requests are asynchronous; Stop drains/revokes audio delivery before + // returning so the connection peer can be destroyed safely. + void StartSpeakerCapturer(); + void StopSpeakerCapturer(); int StartMouseController(); int StopMouseController(); int StartKeyboardCapturer(); @@ -76,8 +86,7 @@ private: SDL_AudioStream *output_stream_ = nullptr; ScreenCapturerFactory *screen_capturer_factory_ = nullptr; ScreenCapturer *screen_capturer_ = nullptr; - SpeakerCapturerFactory *speaker_capturer_factory_ = nullptr; - SpeakerCapturer *speaker_capturer_ = nullptr; + SpeakerCaptureController speaker_capture_; DeviceControllerFactory *device_controller_factory_ = nullptr; MouseController *mouse_controller_ = nullptr; KeyboardCapturer *keyboard_capturer_ = nullptr; @@ -93,4 +102,4 @@ private: } // namespace crossdesk -#endif // CROSSDESK_GUI_SESSION_DEVICE_MANAGER_H_ +#endif \ No newline at end of file diff --git a/apps/desktop/src/gui/runtime/connection_runtime.cpp b/apps/desktop/src/gui/runtime/connection_runtime.cpp index 1519d63a..2733c753 100644 --- a/apps/desktop/src/gui/runtime/connection_runtime.cpp +++ b/apps/desktop/src/gui/runtime/connection_runtime.cpp @@ -168,14 +168,11 @@ void GuiRuntime::HandleServerControllerDisconnected( #else start_screen_capturer_ = false; #endif - start_speaker_capturer_ = false; + devices_.StopSpeakerCapturer(); start_mouse_controller_ = false; start_keyboard_capturer_ = false; remote_client_id_.clear(); - if (audio_capture_) { - devices_.StopSpeakerCapturer(); - audio_capture_ = false; - } + audio_capture_ = false; devices_.ResetToInitialDisplay(); } diff --git a/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp b/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp index b839e50a..de616560 100644 --- a/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp +++ b/apps/desktop/src/gui/runtime/peer_data_callbacks.cpp @@ -298,9 +298,9 @@ void PeerEventHandler::OnReceiveDataBuffer( runtime->devices_.SendMouseCommand(remote_action, runtime->selected_display_); } else if (remote_action.type == ControlType::audio_capture) { - if (remote_action.a && !runtime->start_speaker_capturer_) + if (remote_action.a) runtime->devices_.StartSpeakerCapturer(); - else if (!remote_action.a && runtime->start_speaker_capturer_) + else runtime->devices_.StopSpeakerCapturer(); } else if (remote_action.type == ControlType::keyboard) { runtime->keyboard_.ApplyRemoteEvent(remote_id, remote_action); diff --git a/apps/desktop/src/gui/runtime/peer_event_handler.cpp b/apps/desktop/src/gui/runtime/peer_event_handler.cpp index 67628055..15506997 100644 --- a/apps/desktop/src/gui/runtime/peer_event_handler.cpp +++ b/apps/desktop/src/gui/runtime/peer_event_handler.cpp @@ -468,7 +468,7 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status, std::memory_order_release); runtime->is_server_mode_ = true; runtime->start_screen_capturer_ = true; - runtime->start_speaker_capturer_ = true; + runtime->devices_.StartSpeakerCapturer(); runtime->remote_client_id_ = remote_id; runtime->start_mouse_controller_ = true; { diff --git a/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.cpp b/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.cpp index adb64c1b..02bfa7ae 100644 --- a/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.cpp +++ b/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.cpp @@ -3,269 +3,240 @@ #include #include -#include -#include -#include +#include +#include #include "rd_log.h" namespace crossdesk { - -constexpr int kSampleRate = 48000; -constexpr pa_sample_format_t kFormat = PA_SAMPLE_S16LE; -constexpr int kChannels = 1; +namespace { constexpr size_t kFrameSizeBytes = 480 * sizeof(int16_t); +constexpr auto kStartupTimeout = std::chrono::seconds(2); +constexpr auto kPollInterval = std::chrono::milliseconds(5); -SpeakerCapturerLinux::SpeakerCapturerLinux() - : inited_(false), paused_(false), stop_flag_(false) {} -SpeakerCapturerLinux::~SpeakerCapturerLinux() { - Stop(); - Destroy(); +void ReleaseOperation(pa_operation* operation) { + if (!operation) return; + if (pa_operation_get_state(operation) == PA_OPERATION_RUNNING) + pa_operation_cancel(operation); + pa_operation_unref(operation); } +} // namespace + +SpeakerCapturerLinux::~SpeakerCapturerLinux() { Destroy(); } int SpeakerCapturerLinux::Init(speaker_data_cb cb) { if (inited_) return 0; - cb_ = cb; + cb_ = std::move(cb); inited_ = true; return 0; } -int SpeakerCapturerLinux::Destroy() { - inited_ = false; - return 0; +// Use nonblocking iterations instead of pa_threaded_mainloop_wait(): a server +// that never replies, disappears, or never reaches STREAM_READY must not leave +// startup/Stop waiting for a signal that will never be delivered. +bool SpeakerCapturerLinux::PumpUntil( + const std::function& ready, + std::chrono::steady_clock::time_point deadline) { + while (!stop_flag_) { + if (ready()) return true; + if (std::chrono::steady_clock::now() >= deadline || + !PA_CONTEXT_IS_GOOD(pa_context_get_state(context_))) + return false; + if (pa_mainloop_iterate(mainloop_, 0, nullptr) < 0) return false; + std::this_thread::sleep_for(kPollInterval); + } + return false; } -std::string SpeakerCapturerLinux::GetDefaultMonitorSourceName() { - std::string monitor_name; - std::mutex mtx; - std::condition_variable cv; - bool ready = false; +bool SpeakerCapturerLinux::Prepare() { + const auto deadline = std::chrono::steady_clock::now() + kStartupTimeout; + mainloop_ = pa_mainloop_new(); + if (!mainloop_) return false; + context_ = + pa_context_new(pa_mainloop_get_api(mainloop_), "CrossDesk system audio"); + if (!context_ || + pa_context_connect(context_, nullptr, PA_CONTEXT_NOFLAGS, nullptr) < 0) + return false; + if (!PumpUntil( + [&] { return pa_context_get_state(context_) == PA_CONTEXT_READY; }, + deadline)) + return false; - pa_mainloop* mainloop = pa_mainloop_new(); - pa_mainloop_api* api = pa_mainloop_get_api(mainloop); - pa_context* context = pa_context_new(api, "GetMonitor"); - - struct CallbackState { - std::string* name; - std::mutex* mtx; - std::condition_variable* cv; - bool* ready; - } state{&monitor_name, &mtx, &cv, &ready}; - - pa_context_set_state_callback( - context, - [](pa_context* c, void* userdata) { - auto* state = static_cast(userdata); - if (pa_context_get_state(c) == PA_CONTEXT_READY) { - pa_operation* operation = pa_context_get_server_info( - c, - [](pa_context*, const pa_server_info* info, void* userdata) { - auto* state = static_cast(userdata); - if (info && info->default_sink_name) { - *(state->name) = - std::string(info->default_sink_name) + ".monitor"; - } - { - std::lock_guard lock(*(state->mtx)); - *(state->ready) = true; - } - state->cv->notify_one(); - }, - userdata); - if (operation) { - pa_operation_unref(operation); - } - } + struct ServerResult { + bool done = false; + std::string sink; + } server; + pa_operation* operation = pa_context_get_server_info( + context_, + [](pa_context*, const pa_server_info* info, void* data) { + auto& result = *static_cast(data); + if (info && info->default_sink_name) + result.sink = info->default_sink_name; + result.done = true; }, - &state); + &server); + const bool got_server = + operation && PumpUntil([&] { return server.done; }, deadline); + ReleaseOperation(operation); + if (!got_server || server.sink.empty()) return false; - pa_context_connect(context, nullptr, PA_CONTEXT_NOFLAGS, nullptr); + // Query the actual monitor name; a changed/default device need not follow + // the ".monitor" naming convention. + struct SinkResult { + bool done = false; + std::string monitor; + } sink; + operation = pa_context_get_sink_info_by_name( + context_, server.sink.c_str(), + [](pa_context*, const pa_sink_info* info, int eol, void* data) { + auto& result = *static_cast(data); + if (info && info->monitor_source_name) + result.monitor = info->monitor_source_name; + if (eol != 0) result.done = true; + }, + &sink); + const bool got_sink = + operation && PumpUntil([&] { return sink.done; }, deadline); + ReleaseOperation(operation); + if (!got_sink || sink.monitor.empty()) return false; - std::thread loop_thread([&]() { - int ret = 0; - pa_mainloop_run(mainloop, &ret); - }); + const pa_sample_spec spec{PA_SAMPLE_S16LE, 48000, 1}; + stream_ = pa_stream_new(context_, "System audio", &spec, nullptr); + if (!stream_) return false; + pa_stream_set_read_callback( + stream_, + [](pa_stream* stream, size_t, void* data) { + static_cast(data)->ReadAudio(stream); + }, + this); + pa_buffer_attr attr{}; + attr.maxlength = static_cast(-1); + attr.fragsize = static_cast(kFrameSizeBytes); + if (pa_stream_connect_record(stream_, sink.monitor.c_str(), &attr, + PA_STREAM_ADJUST_LATENCY) < 0) + return false; + if (!PumpUntil( + [&] { + const auto state = pa_stream_get_state(stream_); + return state == PA_STREAM_READY || !PA_STREAM_IS_GOOD(state); + }, + deadline) || + pa_stream_get_state(stream_) != PA_STREAM_READY) + return false; + LOG_INFO("System audio monitor ready: {}", sink.monitor); + return true; +} +void SpeakerCapturerLinux::Run() { + const bool ready = Prepare(); + running_ = ready && !stop_flag_; { - std::unique_lock lock(mtx); - cv.wait_for(lock, std::chrono::seconds(2), [&] { return ready; }); + std::lock_guard lock(state_mtx_); + startup_success_ = running_; + startup_done_ = true; } - - pa_context_disconnect(context); - pa_context_unref(context); - pa_mainloop_quit(mainloop, 0); - loop_thread.join(); - pa_mainloop_free(mainloop); - - return monitor_name; + startup_wake_.notify_one(); + if (!ready && !stop_flag_) { + LOG_ERROR( + "PulseAudio capture startup failed or timed out: {}", + context_ ? pa_strerror(pa_context_errno(context_)) : "no context"); + } + while (running_ && !stop_flag_) { + if (pa_mainloop_iterate(mainloop_, 0, nullptr) < 0 || + pa_context_get_state(context_) != PA_CONTEXT_READY || + pa_stream_get_state(stream_) != PA_STREAM_READY) + break; + std::this_thread::sleep_for(kPollInterval); + } + running_ = false; + Cleanup(); } int SpeakerCapturerLinux::Start() { - if (!inited_ || mainloop_thread_.joinable()) return -1; + if (!inited_) return -1; + if (IsRunning()) return 0; + Stop(); stop_flag_ = false; - - mainloop_thread_ = std::thread([this]() { - std::string monitor_name = GetDefaultMonitorSourceName(); - if (monitor_name.empty()) { - LOG_ERROR("Failed to get monitor source"); - return; - } - - mainloop_ = pa_threaded_mainloop_new(); - pa_mainloop_api* api = pa_threaded_mainloop_get_api(mainloop_); - context_ = pa_context_new(api, "SpeakerCapturer"); - - pa_context_set_state_callback( - context_, - [](pa_context* c, void* userdata) { - auto self = static_cast(userdata); - pa_context_state_t state = pa_context_get_state(c); - if (state == PA_CONTEXT_READY || state == PA_CONTEXT_FAILED || - state == PA_CONTEXT_TERMINATED) { - pa_threaded_mainloop_signal(self->mainloop_, 0); - } - }, - this); - - if (pa_threaded_mainloop_start(mainloop_) < 0) { - LOG_ERROR("Failed to start mainloop"); - Cleanup(); - return; - } - - pa_threaded_mainloop_lock(mainloop_); - - if (pa_context_connect(context_, nullptr, PA_CONTEXT_NOFLAGS, nullptr) < - 0) { - LOG_ERROR("Failed to connect context"); - pa_threaded_mainloop_unlock(mainloop_); - Cleanup(); - return; - } - - while (true) { - pa_context_state_t state = pa_context_get_state(context_); - if (state == PA_CONTEXT_READY) break; - if (!PA_CONTEXT_IS_GOOD(state) || stop_flag_) { - pa_threaded_mainloop_unlock(mainloop_); - Cleanup(); - return; - } - pa_threaded_mainloop_wait(mainloop_); - } - - pa_sample_spec ss = {kFormat, kSampleRate, kChannels}; - stream_ = pa_stream_new(context_, "Capture", &ss, nullptr); - - pa_stream_set_read_callback( - stream_, - [](pa_stream* s, size_t len, void* u) { - auto self = static_cast(u); - if (self->paused_ || self->stop_flag_) return; - - const void* data = nullptr; - if (pa_stream_peek(s, &data, &len) < 0 || !data) return; - - const uint8_t* p = static_cast(data); - self->frame_cache_.insert(self->frame_cache_.end(), p, p + len); - - while (self->frame_cache_.size() >= kFrameSizeBytes) { - std::vector temp_frame( - self->frame_cache_.begin(), - self->frame_cache_.begin() + kFrameSizeBytes); - self->cb_(temp_frame.data(), kFrameSizeBytes, "audio"); - self->frame_cache_.erase( - self->frame_cache_.begin(), - self->frame_cache_.begin() + kFrameSizeBytes); - } - - pa_stream_drop(s); - }, - this); - - pa_buffer_attr attr = {.maxlength = (uint32_t)-1, - .tlength = 0, - .prebuf = 0, - .minreq = 0, - .fragsize = (uint32_t)kFrameSizeBytes}; - - if (pa_stream_connect_record(stream_, monitor_name.c_str(), &attr, - PA_STREAM_ADJUST_LATENCY) < 0) { - LOG_ERROR("Failed to connect stream"); - pa_threaded_mainloop_unlock(mainloop_); - Cleanup(); - return; - } - - while (true) { - pa_stream_state_t s = pa_stream_get_state(stream_); - if (s == PA_STREAM_READY) break; - if (!PA_STREAM_IS_GOOD(s) || stop_flag_) { - pa_threaded_mainloop_unlock(mainloop_); - Cleanup(); - return; - } - pa_threaded_mainloop_wait(mainloop_); - } - - pa_threaded_mainloop_unlock(mainloop_); - - while (!stop_flag_) - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - }); - + paused_ = false; + { + std::lock_guard lock(state_mtx_); + startup_done_ = false; + startup_success_ = false; + } + mainloop_thread_ = std::thread([this] { Run(); }); + std::unique_lock lock(state_mtx_); + const bool started = + startup_wake_.wait_for(lock, kStartupTimeout + std::chrono::seconds(1), + [&] { return startup_done_; }) && + startup_success_; + lock.unlock(); + if (!started) { + Stop(); + return -1; + } return 0; } int SpeakerCapturerLinux::Stop() { stop_flag_ = true; - - if (mainloop_) { - pa_threaded_mainloop_lock(mainloop_); - pa_threaded_mainloop_signal(mainloop_, 0); - pa_threaded_mainloop_unlock(mainloop_); - } - - if (mainloop_thread_.joinable()) { - mainloop_thread_.join(); - } - - Cleanup(); + if (mainloop_thread_.joinable()) mainloop_thread_.join(); + running_ = false; return 0; } +int SpeakerCapturerLinux::Destroy() { + Stop(); + cb_ = nullptr; + inited_ = false; + return 0; +} + +void SpeakerCapturerLinux::ReadAudio(pa_stream* stream) { + const void* data = nullptr; + size_t size = 0; + if (pa_stream_peek(stream, &data, &size) < 0 || size == 0) return; + if (data && !paused_ && !stop_flag_ && cb_) { + const auto* bytes = static_cast(data); + frame_cache_.insert(frame_cache_.end(), bytes, bytes + size); + size_t consumed = 0; + while (!stop_flag_ && frame_cache_.size() - consumed >= kFrameSizeBytes) { + cb_(frame_cache_.data() + consumed, kFrameSizeBytes, "audio"); + consumed += kFrameSizeBytes; + } + frame_cache_.erase(frame_cache_.begin(), frame_cache_.begin() + consumed); + } else { + frame_cache_.clear(); + } + // Drop holes and paused data too, otherwise the read queue cannot advance. + pa_stream_drop(stream); +} + void SpeakerCapturerLinux::Cleanup() { + if (stream_) { + pa_stream_set_read_callback(stream_, nullptr, nullptr); + pa_stream_disconnect(stream_); + pa_stream_unref(stream_); + stream_ = nullptr; + } + if (context_) { + pa_context_disconnect(context_); + pa_context_unref(context_); + context_ = nullptr; + } if (mainloop_) { - pa_threaded_mainloop_stop(mainloop_); - pa_threaded_mainloop_lock(mainloop_); - - if (stream_) { - pa_stream_disconnect(stream_); - pa_stream_unref(stream_); - stream_ = nullptr; - } - - if (context_) { - pa_context_disconnect(context_); - pa_context_unref(context_); - context_ = nullptr; - } - - pa_threaded_mainloop_unlock(mainloop_); - pa_threaded_mainloop_free(mainloop_); + pa_mainloop_free(mainloop_); mainloop_ = nullptr; } - frame_cache_.clear(); } +bool SpeakerCapturerLinux::IsRunning() const { return running_; } int SpeakerCapturerLinux::Pause() { paused_ = true; return 0; } - int SpeakerCapturerLinux::Resume() { paused_ = false; return 0; } -} // namespace crossdesk \ No newline at end of file +} // namespace crossdesk diff --git a/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.h b/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.h index 3f388641..5a9aca67 100644 --- a/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.h +++ b/apps/desktop/src/platform/linux/speaker_capturer/speaker_capturer_linux.h @@ -4,53 +4,59 @@ * Copyright (c) 2025 by DI JUNKUN, All Rights Reserved. */ -#ifndef _SPEAKER_CAPTURER_LINUX_H_ -#define _SPEAKER_CAPTURER_LINUX_H_ +#ifndef CROSSDESK_SPEAKER_CAPTURER_LINUX_H_ +#define CROSSDESK_SPEAKER_CAPTURER_LINUX_H_ #include #include +#include +#include #include #include -#include #include #include #include "speaker_capturer.h" namespace crossdesk { - class SpeakerCapturerLinux : public SpeakerCapturer { public: - SpeakerCapturerLinux(); - ~SpeakerCapturerLinux(); - + SpeakerCapturerLinux() = default; + ~SpeakerCapturerLinux() override; int Init(speaker_data_cb cb) override; int Destroy() override; int Start() override; int Stop() override; - + bool IsRunning() const override; int Pause(); int Resume(); private: - std::string GetDefaultMonitorSourceName(); + void Run(); + bool Prepare(); + bool PumpUntil(const std::function& ready, + std::chrono::steady_clock::time_point deadline); void Cleanup(); + void ReadAudio(pa_stream* stream); - private: - speaker_data_cb cb_ = nullptr; - - std::atomic inited_; - std::atomic paused_; - std::atomic stop_flag_; - + speaker_data_cb cb_; + bool inited_ = false; + std::atomic paused_{false}; + std::atomic stop_flag_{false}; + std::atomic running_{false}; std::thread mainloop_thread_; - pa_threaded_mainloop* mainloop_ = nullptr; + std::mutex state_mtx_; + std::condition_variable startup_wake_; + bool startup_done_ = false; + bool startup_success_ = false; + + // These objects are created, driven and freed exclusively by Run(). Stop + // only changes an atomic flag and joins, never races PulseAudio callbacks. + pa_mainloop* mainloop_ = nullptr; pa_context* context_ = nullptr; pa_stream* stream_ = nullptr; - - std::mutex state_mtx_; std::vector frame_cache_; }; } // namespace crossdesk -#endif \ No newline at end of file +#endif diff --git a/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.h b/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.h index a3093407..653ac3b1 100644 --- a/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.h +++ b/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.h @@ -17,21 +17,19 @@ namespace crossdesk { class SpeakerCapturerMacosx : public SpeakerCapturer { public: SpeakerCapturerMacosx(); - ~SpeakerCapturerMacosx(); + ~SpeakerCapturerMacosx() override; public: - virtual int Init(speaker_data_cb cb); - virtual int Destroy(); - virtual int Start(); - virtual int Stop(); + int Init(speaker_data_cb cb) override; + int Destroy() override; + int Start() override; + int Stop() override; + bool IsRunning() const override; int Pause(); int Resume(); - public: - speaker_data_cb cb_ = nullptr; - bool inited_ = false; - + private: class Impl; Impl* impl_ = nullptr; }; diff --git a/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.mm b/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.mm index 5cea7d85..460e2245 100644 --- a/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.mm +++ b/apps/desktop/src/platform/macos/speaker_capturer/speaker_capturer_macosx.mm @@ -2,47 +2,98 @@ #import #import +#include +#include +#include +#include +#include +#include + #include "rd_log.h" #include "speaker_capturer_macosx.h" -namespace crossdesk { -class SpeakerCapturerMacosx; +namespace { +constexpr auto kCaptureTimeout = std::chrono::seconds(2); + +std::string NSErrorToString(NSError* error) { + if (!error) return ""; + NSString* message = [NSString + stringWithFormat:@"%@ (%@:%ld), reason=%@", error.localizedDescription, error.domain, + (long)error.code, error.localizedFailureReason ?: @""]; + return message.UTF8String ?: ""; } -namespace { -std::string NSErrorToString(NSError* error) { - if (!error) { - return ""; +// Late system replies own only their result, never the capturer or stack data. +// Complete() also tells a late successful start to stop its abandoned stream. +template +class CaptureReply { + public: + bool Complete(T value) { + std::lock_guard lock(mutex_); + if (expired_) return false; + value_ = std::move(value); + ready_ = true; + wake_.notify_one(); + return true; + } + bool Wait(T& value) { + std::unique_lock lock(mutex_); + if (!wake_.wait_for(lock, kCaptureTimeout, [&] { return ready_; })) { + expired_ = true; + return false; + } + value = value_; + return true; } - const char* description = [error.localizedDescription UTF8String]; - return description ? description : ""; -} + private: + std::mutex mutex_; + std::condition_variable wake_; + bool ready_ = false; + bool expired_ = false; + T value_{}; +}; + +struct ContentReply { + SCShareableContent* content = nil; + NSError* error = nil; +}; +struct StreamReply { + NSError* error = nil; +}; +struct AudioCallbackState { + std::mutex mutex; + crossdesk::SpeakerCapturer::speaker_data_cb callback; + std::atomic running{false}; +}; } // namespace -@interface SpeakerCaptureDelegate : NSObject -@property(nonatomic, assign) crossdesk::SpeakerCapturerMacosx* owner; -- (instancetype)initWithOwner:(crossdesk::SpeakerCapturerMacosx*)owner; +@interface SpeakerCaptureDelegate : NSObject { + std::shared_ptr state_; +} +- (instancetype)initWithState:(std::shared_ptr)state; @end @implementation SpeakerCaptureDelegate -- (instancetype)initWithOwner:(crossdesk::SpeakerCapturerMacosx*)owner { +- (instancetype)initWithState:(std::shared_ptr)state { self = [super init]; - if (self) { - _owner = owner; - } + if (self) state_ = std::move(state); return self; } +- (void)stream:(SCStream*)stream didStopWithError:(NSError*)error { + state_->running = false; + SPDLOG_LOGGER_WARN(crossdesk::get_logger(), "System audio stream stopped: {}", + NSErrorToString(error)); +} + - (void)stream:(SCStream*)stream didOutputSampleBuffer:(CMSampleBufferRef)sampleBuffer ofType:(SCStreamOutputType)type { if (type != SCStreamOutputTypeAudio) return; - - crossdesk::SpeakerCapturerMacosx* owner = _owner; - if (!owner || !owner->cb_) { - return; - } + const auto state = state_; + std::lock_guard lock(state->mutex); + if (!state->callback || !state->running) return; CMBlockBufferRef blockBuffer = CMSampleBufferGetDataBuffer(sampleBuffer); if (!blockBuffer) { @@ -51,8 +102,7 @@ std::string NSErrorToString(NSError* error) { size_t length = CMBlockBufferGetDataLength(blockBuffer); char* dataPtr = NULL; - OSStatus dataStatus = - CMBlockBufferGetDataPointer(blockBuffer, 0, NULL, NULL, &dataPtr); + OSStatus dataStatus = CMBlockBufferGetDataPointer(blockBuffer, 0, NULL, NULL, &dataPtr); if (dataStatus != noErr || dataPtr == nullptr || length == 0) { return; } @@ -68,7 +118,7 @@ std::string NSErrorToString(NSError* error) { return; } - if (owner->cb_) { + if (state->callback) { std::vector out_pcm16; if (asbd->mFormatFlags & kAudioFormatFlagIsFloat) { int channels = asbd->mChannelsPerFrame; @@ -118,191 +168,156 @@ std::string NSErrorToString(NSError* error) { size_t total_bytes = out_pcm16.size() * sizeof(short); unsigned char* p = (unsigned char*)out_pcm16.data(); for (size_t offset = 0; offset + frame_bytes <= total_bytes; offset += frame_bytes) { - if (!owner->cb_) { + if (!state->callback) { return; } - owner->cb_(p + offset, frame_bytes, "audio"); + state->callback(p + offset, frame_bytes, "audio"); } } } - @end namespace crossdesk { - class SpeakerCapturerMacosx::Impl { public: - SCStreamConfiguration* config = nil; + speaker_data_cb callback; SCStream* stream = nil; SpeakerCaptureDelegate* delegate = nil; dispatch_queue_t queue = nil; - SCShareableContent* content = nil; - SCDisplay* mainDisplay = nil; - - ~Impl() { - if (stream) { - [stream stopCaptureWithCompletionHandler:^(NSError* _Nullable error){ - }]; - stream = nil; - } - delegate = nil; - if (queue) { - queue = nil; - } - content = nil; - mainDisplay = nil; - config = nil; - } + std::shared_ptr state; }; -SpeakerCapturerMacosx::SpeakerCapturerMacosx() { - impl_ = new Impl(); - inited_ = false; - cb_ = nullptr; -} - +SpeakerCapturerMacosx::SpeakerCapturerMacosx() : impl_(new Impl) {} SpeakerCapturerMacosx::~SpeakerCapturerMacosx() { Destroy(); delete impl_; - impl_ = nullptr; } int SpeakerCapturerMacosx::Init(speaker_data_cb cb) { - if (inited_) { - return 0; - } - cb_ = cb; - - impl_->config = [[SCStreamConfiguration alloc] init]; - impl_->config.capturesAudio = YES; - impl_->config.sampleRate = 48000; - impl_->config.channelCount = 1; - - dispatch_semaphore_t sema = dispatch_semaphore_create(0); - __block NSError* error = nil; - [SCShareableContent - getShareableContentWithCompletionHandler:^(SCShareableContent* c, NSError* e) { - impl_->content = c; - error = e; - dispatch_semaphore_signal(sema); - }]; - dispatch_semaphore_wait(sema, DISPATCH_TIME_FOREVER); - - if (error || !impl_->content) { - LOG_ERROR("Failed to get shareable content: {}", - NSErrorToString(error)); - return -1; - } - - CGDirectDisplayID mainDisplayId = CGMainDisplayID(); - impl_->mainDisplay = nil; - for (SCDisplay* d in impl_->content.displays) { - if (d.displayID == mainDisplayId) { - impl_->mainDisplay = d; - break; - } - } - if (!impl_->mainDisplay) { - LOG_ERROR("Main display not found"); - return -1; - } - - if (!impl_->queue) { + impl_->callback = std::move(cb); + if (!impl_->queue) impl_->queue = dispatch_queue_create("SpeakerAudio.Queue", DISPATCH_QUEUE_SERIAL); - } - - inited_ = true; return 0; } int SpeakerCapturerMacosx::Start() { - if (!inited_) { - return -1; - } + @autoreleasepool { + if (!impl_->callback) return -1; + if (IsRunning()) return 0; + if (Stop() != 0) return -1; - if (impl_->stream) { - dispatch_semaphore_t semaStop = dispatch_semaphore_create(0); - [impl_->stream stopCaptureWithCompletionHandler:^(NSError* error) { - dispatch_semaphore_signal(semaStop); - }]; - dispatch_semaphore_wait(semaStop, DISPATCH_TIME_FOREVER); - impl_->stream = nil; - impl_->delegate = nil; - } - - impl_->delegate = [[SpeakerCaptureDelegate alloc] initWithOwner:this]; - SCContentFilter* filter = [[SCContentFilter alloc] initWithDisplay:impl_->mainDisplay - excludingWindows:@[]]; - impl_->stream = [[SCStream alloc] initWithFilter:filter - configuration:impl_->config - delegate:impl_->delegate]; - - NSError* addOutputError = nil; - BOOL ok = [impl_->stream addStreamOutput:impl_->delegate - type:SCStreamOutputTypeAudio - sampleHandlerQueue:impl_->queue - error:&addOutputError]; - if (!ok || addOutputError) { - LOG_ERROR("addStreamOutput error: {}", - NSErrorToString(addOutputError)); - impl_->stream = nil; - impl_->delegate = nil; - return -1; - } - - dispatch_semaphore_t semaStart = dispatch_semaphore_create(0); - __block int ret = 0; - [impl_->stream startCaptureWithCompletionHandler:^(NSError* _Nullable error) { - if (error) { - LOG_ERROR("startCaptureWithCompletionHandler error: {}", - NSErrorToString(error)); - ret = -1; + // A cached SCDisplay becomes invalid after hotplug, sleep or a display + // reconfiguration. Obtain a fresh snapshot on every start/reconnection. + auto content_reply = std::make_shared>(); + [SCShareableContent + getShareableContentWithCompletionHandler:^(SCShareableContent* content, NSError* error) { + content_reply->Complete({content, error}); + }]; + ContentReply content; + if (!content_reply->Wait(content)) { + LOG_ERROR("Timed out refreshing system audio capture displays"); + return -1; } - dispatch_semaphore_signal(semaStart); - }]; - dispatch_semaphore_wait(semaStart, DISPATCH_TIME_FOREVER); + if (content.error || !content.content) { + LOG_ERROR("Failed to refresh system audio capture displays: {}", + NSErrorToString(content.error)); + return -1; + } + SCDisplay* display = nil; + const auto main_id = CGMainDisplayID(); + for (SCDisplay* candidate in content.content.displays) { + if (candidate.displayID == main_id) { + display = candidate; + break; + } + } + if (!display) display = content.content.displays.firstObject; + if (!display) { + LOG_ERROR("No display available for system audio capture"); + return -1; + } + LOG_INFO("Starting system audio capture with current display {}", display.displayID); - return ret; + SCStreamConfiguration* config = [[SCStreamConfiguration alloc] init]; + config.capturesAudio = YES; + config.sampleRate = 48000; + config.channelCount = 1; + SCContentFilter* filter = [[SCContentFilter alloc] initWithDisplay:display + excludingWindows:@[]]; + impl_->state = std::make_shared(); + impl_->state->callback = impl_->callback; + impl_->delegate = [[SpeakerCaptureDelegate alloc] initWithState:impl_->state]; + impl_->stream = [[SCStream alloc] initWithFilter:filter + configuration:config + delegate:impl_->delegate]; + NSError* output_error = nil; + if (!impl_->stream || ![impl_->stream addStreamOutput:impl_->delegate + type:SCStreamOutputTypeAudio + sampleHandlerQueue:impl_->queue + error:&output_error]) { + LOG_ERROR("Failed to attach system audio output: {}", NSErrorToString(output_error)); + Stop(); + return -1; + } + + auto start_reply = std::make_shared>(); + SCStream* stream = impl_->stream; + impl_->state->running = true; + [stream startCaptureWithCompletionHandler:^(NSError* error) { + if (!start_reply->Complete({error}) && !error) { + [stream stopCaptureWithCompletionHandler:nil]; + } + }]; + StreamReply started; + if (!start_reply->Wait(started)) { + LOG_ERROR("Timed out starting system audio capture"); + Stop(); + return -1; + } + if (started.error || !impl_->state->running) { + LOG_ERROR("Failed to start system audio capture: {}", NSErrorToString(started.error)); + Stop(); + return -1; + } + return 0; + } } int SpeakerCapturerMacosx::Stop() { - if (!inited_) return -1; - if (!impl_->stream) return -1; - - dispatch_semaphore_t sema = dispatch_semaphore_create(0); - [impl_->stream stopCaptureWithCompletionHandler:^(NSError* error) { - if (error) { - LOG_ERROR("stopCaptureWithCompletionHandler error: {}", - NSErrorToString(error)); + @autoreleasepool { + if (auto state = std::exchange(impl_->state, {})) { + std::lock_guard lock(state->mutex); + state->running = false; + state->callback = nullptr; } - dispatch_semaphore_signal(sema); - }]; - dispatch_semaphore_wait(sema, DISPATCH_TIME_FOREVER); - - impl_->stream = nil; - impl_->delegate.owner = nullptr; - impl_->delegate = nil; - - return 0; + SCStream* stream = impl_->stream; + impl_->stream = nil; + impl_->delegate = nil; + if (!stream) return 0; + auto stop_reply = std::make_shared>(); + [stream stopCaptureWithCompletionHandler:^(NSError* error) { + stop_reply->Complete({error}); + }]; + StreamReply stopped; + if (!stop_reply->Wait(stopped)) { + LOG_ERROR("Timed out stopping system audio capture; callbacks revoked"); + return -1; + } + // A start failure or daemon restart can leave an already stopped stream. + // Its references have been discarded, so future starts can recover. + if (stopped.error) LOG_WARN("System audio stop reply: {}", NSErrorToString(stopped.error)); + return 0; + } } +bool SpeakerCapturerMacosx::IsRunning() const { return impl_->state && impl_->state->running; } + int SpeakerCapturerMacosx::Destroy() { - Stop(); - cb_ = nullptr; - - if (impl_) { - impl_->config = nil; - impl_->content = nil; - impl_->mainDisplay = nil; - if (impl_->queue) { - impl_->queue = nil; - } - } - inited_ = false; - return 0; + const int result = Stop(); + impl_->callback = nullptr; + impl_->queue = nil; + return result; } - -int SpeakerCapturerMacosx::Pause() { return 0; } - +int SpeakerCapturerMacosx::Pause() { return Stop(); } int SpeakerCapturerMacosx::Resume() { return Start(); } } // namespace crossdesk diff --git a/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.cpp b/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.cpp index 401496ce..4e810a42 100644 --- a/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.cpp +++ b/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.cpp @@ -1,104 +1,95 @@ #include "speaker_capturer_wasapi.h" +#include + #include "rd_log.h" #define MINIAUDIO_IMPLEMENTATION #include "miniaudio.h" -#define SAVE_AUDIO_FILE 0 - namespace crossdesk { -static ma_device_config device_config_; -static ma_device device_; -static ma_format format_ = ma_format_s16; -static ma_uint32 sample_rate_ = ma_standard_sample_rate_48000; -static ma_uint32 channels_ = 1; -static FILE* fp_ = nullptr; - -void data_callback(ma_device* pDevice, void* pOutput, const void* pInput, - ma_uint32 frameCount) { - SpeakerCapturerWasapi* ptr = (SpeakerCapturerWasapi*)pDevice->pUserData; - if (ptr) { - if (SAVE_AUDIO_FILE) { - fwrite(pInput, frameCount * ma_get_bytes_per_frame(format_, channels_), 1, - fp_); - } - - ptr->GetCallback()((unsigned char*)pInput, - frameCount * ma_get_bytes_per_frame(format_, channels_), - "audio"); - } - - (void)pOutput; -} - -SpeakerCapturerWasapi::speaker_data_cb SpeakerCapturerWasapi::GetCallback() { - return cb_; -} - -SpeakerCapturerWasapi::SpeakerCapturerWasapi() {} +// Each attempt owns its device. A failed/old session must never uninitialize +// a device belonging to another capturer or reuse already released storage. +class SpeakerCapturerWasapi::Impl { + public: + ma_device device{}; + bool initialized = false; + speaker_data_cb callback; +}; +SpeakerCapturerWasapi::SpeakerCapturerWasapi() : impl_(new Impl) {} SpeakerCapturerWasapi::~SpeakerCapturerWasapi() { - if (SAVE_AUDIO_FILE) { - fclose(fp_); - } + Destroy(); + delete impl_; } int SpeakerCapturerWasapi::Init(speaker_data_cb cb) { - if (inited_) { - return 0; - } - - cb_ = cb; - - if (SAVE_AUDIO_FILE) { - fopen_s(&fp_, "system_audio.pcm", "wb"); - } - - ma_result result; + if (impl_->initialized) return 0; + impl_->callback = std::move(cb); + ma_device_config config = ma_device_config_init(ma_device_type_loopback); + config.capture.pDeviceID = + nullptr; // Resolve the current default each attempt. + config.capture.format = ma_format_s16; + config.capture.channels = 1; + config.sampleRate = 48000; + config.pUserData = impl_; + config.dataCallback = [](ma_device* device, void*, const void* input, + ma_uint32 frames) { + auto* state = static_cast(device->pUserData); + if (state && state->callback && input && frames > 0) + state->callback((unsigned char*)input, frames * sizeof(int16_t), "audio"); + }; ma_backend backends[] = {ma_backend_wasapi}; - - device_config_ = ma_device_config_init(ma_device_type_loopback); - device_config_.capture.pDeviceID = NULL; - device_config_.capture.format = format_; - device_config_.capture.channels = channels_; - device_config_.sampleRate = sample_rate_; - device_config_.dataCallback = data_callback; - device_config_.pUserData = this; - - result = ma_device_init_ex(backends, sizeof(backends) / sizeof(backends[0]), - NULL, &device_config_, &device_); + const ma_result result = + ma_device_init_ex(backends, 1, nullptr, &config, &impl_->device); if (result != MA_SUCCESS) { - LOG_ERROR("Failed to initialize loopback device"); + impl_->callback = nullptr; + LOG_ERROR("Failed to initialize WASAPI loopback device: {}", + static_cast(result)); return -1; } - - inited_ = true; - + impl_->initialized = true; return 0; } int SpeakerCapturerWasapi::Start() { - ma_result result = ma_device_start(&device_); + if (!impl_->initialized) return -1; + if (IsRunning()) return 0; + const ma_result result = ma_device_start(&impl_->device); if (result != MA_SUCCESS) { - ma_device_uninit(&device_); - LOG_ERROR("Failed to start device"); + LOG_ERROR("Failed to start WASAPI loopback device: {}", + static_cast(result)); + Destroy(); return -1; } - return 0; } int SpeakerCapturerWasapi::Stop() { - ma_device_stop(&device_); + if (!impl_->initialized || !IsRunning()) return 0; + const ma_result result = ma_device_stop(&impl_->device); + if (result != MA_SUCCESS) { + LOG_WARN("Failed to stop WASAPI loopback device: {}", + static_cast(result)); + return -1; + } return 0; } int SpeakerCapturerWasapi::Destroy() { - ma_device_uninit(&device_); + if (impl_->initialized) { + // ma_device_uninit also stops and drains the device callback thread. + ma_device_uninit(&impl_->device); + impl_->initialized = false; + } + impl_->callback = nullptr; return 0; } -int SpeakerCapturerWasapi::Pause() { return 0; } -} // namespace crossdesk \ No newline at end of file +bool SpeakerCapturerWasapi::IsRunning() const { + return impl_->initialized && ma_device_is_started(&impl_->device); +} +int SpeakerCapturerWasapi::Pause() { return Stop(); } +int SpeakerCapturerWasapi::Resume() { return Start(); } +} // namespace crossdesk diff --git a/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.h b/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.h index cbc667e2..5ebd653e 100644 --- a/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.h +++ b/apps/desktop/src/platform/windows/speaker_capturer/speaker_capturer_wasapi.h @@ -14,24 +14,21 @@ namespace crossdesk { class SpeakerCapturerWasapi : public SpeakerCapturer { public: SpeakerCapturerWasapi(); - ~SpeakerCapturerWasapi(); + ~SpeakerCapturerWasapi() override; public: - virtual int Init(speaker_data_cb cb); - virtual int Destroy(); - virtual int Start(); - virtual int Stop(); + int Init(speaker_data_cb cb) override; + int Destroy() override; + int Start() override; + int Stop() override; + bool IsRunning() const override; int Pause(); int Resume(); - speaker_data_cb GetCallback(); - private: - speaker_data_cb cb_ = nullptr; - - private: - bool inited_ = false; + class Impl; + Impl* impl_ = nullptr; }; } // namespace crossdesk #endif \ No newline at end of file diff --git a/apps/desktop/src/speaker_capturer/speaker_capture_controller.cpp b/apps/desktop/src/speaker_capturer/speaker_capture_controller.cpp new file mode 100644 index 00000000..d7fba1dc --- /dev/null +++ b/apps/desktop/src/speaker_capturer/speaker_capture_controller.cpp @@ -0,0 +1,165 @@ +#include "speaker_capture_controller.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "rd_log.h" + +namespace crossdesk { + +struct SpeakerCaptureController::State { + Factory factory; + RetryPolicy policy; + std::mutex mutex; + std::condition_variable wake; + bool enabled = false; + bool shutdown = false; + uint64_t generation = 0; + std::atomic running{false}; + + // Never hold this mutex across a native capture operation. It only drains + // an in-flight audio delivery when a session is disabled/destroyed. + std::mutex callback_mutex; + SpeakerCapturer::speaker_data_cb callback; + bool deliver = false; + uint64_t callback_attempt = 0; +}; + +SpeakerCaptureController::SpeakerCaptureController( + Factory factory, SpeakerCapturer::speaker_data_cb cb) + : SpeakerCaptureController(std::move(factory), std::move(cb), + RetryPolicy{}) {} + +SpeakerCaptureController::SpeakerCaptureController( + Factory factory, SpeakerCapturer::speaker_data_cb cb, RetryPolicy policy) + : state_(std::make_shared()) { + state_->factory = std::move(factory); + state_->callback = std::move(cb); + policy.max_attempts = std::max(1u, policy.max_attempts); + state_->policy = policy; + // The worker owns only shared state, never this/controller/GUI. Even a + // driver stuck inside a native API cannot block shutdown or outlive an + // unguarded callback target. There is at most one worker per controller. + std::thread([state = state_] { Run(state); }).detach(); +} + +SpeakerCaptureController::~SpeakerCaptureController() { Shutdown(); } + +void SpeakerCaptureController::SetEnabled(bool enabled) { + std::lock_guard lock(state_->mutex); + if (state_->shutdown || state_->enabled == enabled) return; + state_->enabled = enabled; + ++state_->generation; + if (!enabled) { + std::lock_guard callback_lock(state_->callback_mutex); + state_->deliver = false; + state_->running = false; + } + state_->wake.notify_one(); +} + +bool SpeakerCaptureController::IsRunning() const { return state_->running; } + +void SpeakerCaptureController::Shutdown() { + std::lock_guard lock(state_->mutex); + state_->shutdown = true; + state_->enabled = false; + state_->running = false; + { + std::lock_guard callback_lock(state_->callback_mutex); + state_->deliver = false; + state_->callback = nullptr; + } + state_->wake.notify_one(); +} + +void SpeakerCaptureController::Run(const std::shared_ptr& state) { + uint64_t callback_attempt = 0; + std::unique_lock lock(state->mutex); + while (!state->shutdown) { + state->wake.wait(lock, [&] { return state->shutdown || state->enabled; }); + if (state->shutdown) break; + const uint64_t generation = state->generation; + const auto cancelled = [&] { + return state->shutdown || !state->enabled || + state->generation != generation; + }; + + for (unsigned attempt = 1; !cancelled(); ++attempt) { + const uint64_t current_attempt = ++callback_attempt; + lock.unlock(); + auto capturer = state->factory(); + std::weak_ptr weak = state; + auto callback = [weak, current_attempt](unsigned char* data, size_t size, + const char* name) { + if (auto current = weak.lock()) { + std::lock_guard callback_lock(current->callback_mutex); + if (current->deliver && + current->callback_attempt == current_attempt && + current->callback) { + current->callback(data, size, name); + } + } + }; + const bool initialized = capturer && capturer->Init(callback) == 0; + lock.lock(); + const bool should_start = initialized && !cancelled(); + lock.unlock(); + const bool started = should_start && capturer->Start() == 0; + lock.lock(); + if (started && !cancelled()) { + { + std::lock_guard callback_lock(state->callback_mutex); + state->callback_attempt = current_attempt; + state->deliver = true; + state->running = true; + } + LOG_INFO("System audio capture started (attempt {})", attempt); + while (!cancelled()) { + if (state->wake.wait_for(lock, state->policy.health_interval, + cancelled)) + break; + lock.unlock(); + const bool healthy = capturer->IsRunning(); + lock.lock(); + if (!healthy) { + LOG_WARN("System audio capture stopped unexpectedly"); + break; + } + } + } + { + std::lock_guard callback_lock(state->callback_mutex); + state->deliver = false; + state->running = false; + } + lock.unlock(); + // Recreate on every attempt/reconnection to discard stale displays, + // PulseAudio sources and invalidated WASAPI devices. + if (capturer) capturer->Destroy(); + capturer.reset(); + lock.lock(); + if (cancelled()) break; + if (attempt >= state->policy.max_attempts) { + LOG_ERROR( + "System audio capture disabled after {} attempts; toggle audio " + "or reconnect to retry", + attempt); + state->wake.wait(lock, cancelled); + break; + } + const auto delay = + state->policy.initial_delay * (1u << std::min(attempt - 1, 5u)); + LOG_WARN("System audio capture failed (attempt {}); retry in {} ms", + attempt, delay.count()); + state->wake.wait_for(lock, delay, cancelled); + } + } +} + +} // namespace crossdesk diff --git a/apps/desktop/src/speaker_capturer/speaker_capture_controller.h b/apps/desktop/src/speaker_capturer/speaker_capture_controller.h new file mode 100644 index 00000000..473fa16d --- /dev/null +++ b/apps/desktop/src/speaker_capturer/speaker_capture_controller.h @@ -0,0 +1,50 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-09-09 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _SPEAKER_CAPTURE_CONTROLLER_H_ +#define _SPEAKER_CAPTURE_CONTROLLER_H_ + +#include +#include +#include + +#include "speaker_capturer.h" + +namespace crossdesk { + +// Serializes platform capture operations away from the UI and transport +// threads. Stop/Shutdown revoke callbacks before returning; native teardown +// can then finish without keeping the GUI or its peer alive. +class SpeakerCaptureController { + public: + using Factory = std::function()>; + struct RetryPolicy { + unsigned max_attempts = 3; + std::chrono::milliseconds initial_delay{1000}; + std::chrono::milliseconds health_interval{100}; + }; + + SpeakerCaptureController(Factory factory, + SpeakerCapturer::speaker_data_cb cb); + SpeakerCaptureController(Factory factory, SpeakerCapturer::speaker_data_cb cb, + RetryPolicy policy); + ~SpeakerCaptureController(); + SpeakerCaptureController(const SpeakerCaptureController&) = delete; + SpeakerCaptureController& operator=(const SpeakerCaptureController&) = delete; + + void SetEnabled(bool enabled); + bool IsRunning() const; + void Shutdown(); + + private: + struct State; + static void Run(const std::shared_ptr& state); + std::shared_ptr state_; +}; + +} // namespace crossdesk + +#endif \ No newline at end of file diff --git a/apps/desktop/src/speaker_capturer/speaker_capturer.h b/apps/desktop/src/speaker_capturer/speaker_capturer.h index 5ceb4a0e..c9255ab7 100644 --- a/apps/desktop/src/speaker_capturer/speaker_capturer.h +++ b/apps/desktop/src/speaker_capturer/speaker_capturer.h @@ -24,6 +24,8 @@ class SpeakerCapturer { virtual int Destroy() = 0; virtual int Start() = 0; virtual int Stop() = 0; + // Called on the same control worker as Init/Start/Stop. Must not block. + virtual bool IsRunning() const = 0; }; } // namespace crossdesk -#endif \ No newline at end of file +#endif diff --git a/apps/desktop/xmake/targets.lua b/apps/desktop/xmake/targets.lua index 79a1522a..05a7b0ce 100644 --- a/apps/desktop/xmake/targets.lua +++ b/apps/desktop/xmake/targets.lua @@ -193,13 +193,15 @@ function setup_targets() set_kind("object") add_deps("rd_log", "crossdesk_wire") add_includedirs("apps/desktop/src/speaker_capturer", {public = true}) + add_files("apps/desktop/src/speaker_capturer/speaker_capture_controller.cpp") if is_os("windows") then add_packages("miniaudio") add_files("apps/desktop/src/platform/windows/speaker_capturer/*.cpp") add_includedirs("apps/desktop/src/platform/windows/speaker_capturer") elseif is_os("macosx") then - add_files("apps/desktop/src/platform/macos/speaker_capturer/*.cpp", - "apps/desktop/src/platform/macos/speaker_capturer/*.mm") + add_files("apps/desktop/src/platform/macos/speaker_capturer/*.cpp") + add_files("apps/desktop/src/platform/macos/speaker_capturer/*.mm", + {mxxflags = "-fobjc-arc"}) add_includedirs("apps/desktop/src/platform/macos/speaker_capturer") elseif is_os("linux") then add_files("apps/desktop/src/platform/linux/speaker_capturer/*.cpp")