ladybird/Libraries/LibMedia/Audio/PlaybackStreamPulseAudio.cpp

/*
 * Copyright (c) 2023-2026, Gregory Bertilson <gregory@ladybird.org>
 *
 * SPDX-License-Identifier: BSD-2-Clause
 */

#include <LibCore/EventLoop.h>
#include <LibCore/ThreadedPromise.h>
#include <LibThreading/Thread.h>

#include "PlaybackStreamPulseAudio.h"

namespace Audio {

#define TRY_OR_REJECT_AND_EXIT(expression)                                                                                         \
    ({                                                                                                                             \
        auto&& __temporary_result = (expression);                                                                                  \
        if (__temporary_result.is_error()) [[unlikely]] {                                                                          \
            warnln("Failure in PulseAudio control thread: {}", __temporary_result.error().string_literal());                       \
            main_thread_event_loop.deferred_invoke([promise = move(promise), error = __temporary_result.release_error()] mutable { \
                promise->reject(move(error));                                                                                      \
            });                                                                                                                    \
            internal_state->exit();                                                                                                \
            return 1;                                                                                                              \
        }                                                                                                                          \
        __temporary_result.release_value();                                                                                        \
    })

NonnullRefPtr<PlaybackStream::CreatePromise> PlaybackStream::create(OutputState initial_output_state, u32 target_latency_ms, AudioDataRequestCallback&& data_request_callback)
{
    return PlaybackStreamPulseAudio::create(initial_output_state, target_latency_ms, move(data_request_callback));
}

NonnullRefPtr<PlaybackStream::CreatePromise> PlaybackStreamPulseAudio::create(OutputState initial_state, u32 target_latency_ms, AudioDataRequestCallback&& data_request_callback)
{
    VERIFY(data_request_callback);

    auto promise = CreatePromise::construct();

    // Create an internal state for the control thread to hold on to.
    auto internal_state = MUST(adopt_nonnull_ref_or_enomem(new (nothrow) InternalState()));
    auto playback_stream = MUST(adopt_nonnull_ref_or_enomem(new (nothrow) PlaybackStreamPulseAudio(internal_state)));
    auto& main_thread_event_loop = Core::EventLoop::current();

    // Create the control thread and start it.
    auto thread = MUST(Threading::Thread::try_create("Audio Control"sv, [=, &main_thread_event_loop, data_request_callback = move(data_request_callback)]() mutable {
        auto context = TRY_OR_REJECT_AND_EXIT(PulseAudioContext::the());
        internal_state->set_stream(TRY_OR_REJECT_AND_EXIT(context->create_stream(initial_state, target_latency_ms, [data_request_callback = move(data_request_callback)](PulseAudioStream&, Span<float> buffer) {
            return data_request_callback(buffer);
        })));

        // PulseAudio retains the last volume it sets for an application. We want to consistently
        // start at 100% volume instead.
        TRY_OR_REJECT_AND_EXIT(internal_state->stream()->set_volume(1.0));

        {
            main_thread_event_loop.deferred_invoke([promise = move(promise), playback_stream = move(playback_stream)] mutable {
                promise->resolve(move(playback_stream));
            });
        }

        internal_state->thread_loop();
        return 0;
    }));

    thread->start();
    thread->detach();
    return promise;
}

PlaybackStreamPulseAudio::PlaybackStreamPulseAudio(NonnullRefPtr<InternalState> state)
    : m_state(move(state))
{
}

SampleSpecification PlaybackStreamPulseAudio::sample_specification() const
{
    return m_state->stream()->sample_specification();
}

PlaybackStreamPulseAudio::~PlaybackStreamPulseAudio()
{
    m_state->exit();
}

#define TRY_OR_REJECT(expression, ...)                           \
    ({                                                           \
        auto&& __temporary_result = (expression);                \
        if (__temporary_result.is_error()) [[unlikely]] {        \
            promise->reject(__temporary_result.release_error()); \
            return __VA_ARGS__;                                  \
        }                                                        \
        __temporary_result.release_value();                      \
    })

void PlaybackStreamPulseAudio::set_underrun_callback(Function<void()> callback)
{
    m_state->enqueue([&state = *m_state, callback = move(callback)]() mutable {
        state.stream()->set_underrun_callback(move(callback));
    });
}

NonnullRefPtr<Core::ThreadedPromise<AK::Duration>> PlaybackStreamPulseAudio::resume()
{
    auto promise = Core::ThreadedPromise<AK::Duration>::create();
    TRY_OR_REJECT(m_state->check_is_running(), promise);
    m_state->enqueue([&state = *m_state, promise]() {
        TRY_OR_REJECT(state.stream()->resume());
        promise->resolve(state.stream()->total_time_played());
    });
    return promise;
}

NonnullRefPtr<Core::ThreadedPromise<void>> PlaybackStreamPulseAudio::drain_buffer_and_suspend()
{
    auto promise = Core::ThreadedPromise<void>::create();
    TRY_OR_REJECT(m_state->check_is_running(), promise);
    m_state->enqueue([&state = *m_state, promise]() {
        TRY_OR_REJECT(state.stream()->drain_and_suspend());
        promise->resolve();
    });
    return promise;
}

NonnullRefPtr<Core::ThreadedPromise<void>> PlaybackStreamPulseAudio::discard_buffer_and_suspend()
{
    auto promise = Core::ThreadedPromise<void>::create();
    TRY_OR_REJECT(m_state->check_is_running(), promise);
    m_state->enqueue([&state = *m_state, promise]() {
        TRY_OR_REJECT(state.stream()->flush_and_suspend());
        promise->resolve();
    });
    return promise;
}

void PlaybackStreamPulseAudio::notify_data_available()
{
    if (m_state->stream() != nullptr)
        m_state->stream()->notify_data_available();
}

AK::Duration PlaybackStreamPulseAudio::total_time_played() const
{
    if (m_state->stream() != nullptr)
        return m_state->stream()->total_time_played();
    return AK::Duration::zero();
}

NonnullRefPtr<Core::ThreadedPromise<void>> PlaybackStreamPulseAudio::set_volume(double volume)
{
    auto promise = Core::ThreadedPromise<void>::create();
    TRY_OR_REJECT(m_state->check_is_running(), promise);
    m_state->enqueue([&state = *m_state, promise, volume]() {
        TRY_OR_REJECT(state.stream()->set_volume(volume));
        promise->resolve();
    });
    return promise;
}

ErrorOr<void> PlaybackStreamPulseAudio::InternalState::check_is_running()
{
    if (m_exit)
        return Error::from_string_literal("PulseAudio control thread loop is not running");
    return {};
}

void PlaybackStreamPulseAudio::InternalState::set_stream(NonnullRefPtr<PulseAudioStream>&& stream)
{
    m_stream = move(stream);
}

RefPtr<PulseAudioStream> const& PlaybackStreamPulseAudio::InternalState::stream()
{
    return m_stream;
}

void PlaybackStreamPulseAudio::InternalState::enqueue(Function<void()>&& task)
{
    Sync::MutexLocker locker { m_mutex };
    m_tasks.enqueue(forward<Function<void()>>(task));
    m_wake_condition.signal();
}

void PlaybackStreamPulseAudio::InternalState::thread_loop()
{
    while (true) {
        auto task = [this]() -> Function<void()> {
            Sync::MutexLocker locker { m_mutex };

            while (m_tasks.is_empty() && !m_exit)
                m_wake_condition.wait();
            if (m_exit)
                return nullptr;
            return m_tasks.dequeue();
        }();
        if (!task) {
            VERIFY(m_exit);
            break;
        }
        task();
    }
}

void PlaybackStreamPulseAudio::InternalState::exit()
{
    m_exit = true;
    m_wake_condition.signal();
}

}