femtosynth/pipemini/pipemini.cpp

#include "pipemini.h"
#include <thread>
#include <atomic>
#include <mutex>
#include <condition_variable>

#define DEFAULT_CHANNELS 2
#define DEFAULT_FORMAT SPA_AUDIO_FORMAT_S16

struct PipeMiniData {
    PipeMini* self;
    std::thread loop_thread;
    std::atomic<bool> should_quit{false};
    std::mutex mtx;
    std::condition_variable cv;
    bool teardown_done = false;
};

PipeMini::PipeMini() = default;

PipeMini::~PipeMini() {
    stop();
}

void PipeMini::teardown() {
    if (stream_) {
        pw_stream_disconnect(stream_);
        pw_stream_destroy(stream_);
        stream_ = nullptr;
    }

    spa_hook_remove(&stream_listener_);

    if (core_) {
        pw_core_disconnect(core_);
        core_ = nullptr;
    }

    if (context_) {
        pw_context_destroy(context_);
        context_ = nullptr;
    }

    if (loop_) {
        pw_main_loop_quit(loop_);
    }
}

static int on_teardown_event(spa_loop* loop, bool async, uint32_t id, const void* data, size_t size, void* userdata) {
    (void)loop;
    (void)async;
    (void)id;
    (void)data;
    (void)size;

    auto* pmd = static_cast<PipeMiniData*>(userdata);
    PipeMini* self = pmd->self;

    self->teardown();

    {
        std::lock_guard<std::mutex> lock(pmd->mtx);
        pmd->teardown_done = true;
    }
    pmd->cv.notify_one();

    return 0;
}

static const pw_stream_events stream_events = {
    PW_VERSION_STREAM_EVENTS,
    .process = PipeMini::on_process,
};

bool PipeMini::start(const Callback& cb, int sample_rate) {
    if (running_) return false;

    callback_ = cb;
    sample_rate_ = sample_rate;

    pw_init(nullptr, nullptr);

    loop_ = pw_main_loop_new(nullptr);
    if (!loop_) return false;

    context_ = pw_context_new(pw_main_loop_get_loop(loop_), nullptr, 0);
    if (!context_) {
        pw_main_loop_destroy(loop_);
        loop_ = nullptr;
        return false;
    }

    core_ = pw_context_connect(context_, nullptr, 0);
    if (!core_) {
        pw_context_destroy(context_);
        context_ = nullptr;
        pw_main_loop_destroy(loop_);
        loop_ = nullptr;
        return false;
    }

    stream_ = pw_stream_new(
        core_,
        "pipemini",
        pw_properties_new(
            PW_KEY_MEDIA_TYPE, "Audio",
            PW_KEY_MEDIA_CATEGORY, "Playback",
            PW_KEY_MEDIA_ROLE, "Music",
            nullptr));
    if (!stream_) {
        pw_main_loop_destroy(loop_);
        loop_ = nullptr;
        return false;
    }

    pw_stream_add_listener(stream_, &stream_listener_, &stream_events, this);

    uint8_t buffer[1024];
    spa_pod_builder b = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));

    spa_audio_info_raw info = {};
    info.format = DEFAULT_FORMAT;
    info.channels = DEFAULT_CHANNELS;
    info.rate = (uint32_t)sample_rate_;
    info.position[0] = SPA_AUDIO_CHANNEL_FL;
    info.position[1] = SPA_AUDIO_CHANNEL_FR;

    const spa_pod* params = spa_format_audio_raw_build(&b, SPA_PARAM_EnumFormat, &info);

    if (pw_stream_connect(stream_,
                          PW_DIRECTION_OUTPUT,
                          PW_ID_ANY,
                          (pw_stream_flags)(PW_STREAM_FLAG_AUTOCONNECT |
                                            PW_STREAM_FLAG_MAP_BUFFERS |
                                            PW_STREAM_FLAG_RT_PROCESS),
                          &params, 1) < 0) {
        pw_stream_destroy(stream_);
        stream_ = nullptr;
        pw_core_disconnect(core_);
        core_ = nullptr;
        pw_context_destroy(context_);
        context_ = nullptr;
        pw_main_loop_destroy(loop_);
        loop_ = nullptr;
        return false;
    }

    running_ = true;

    // Run the PipeWire loop in a background thread.
    data_ = new PipeMiniData{this, {}, {}};
    data_->loop_thread = std::thread([this]() {
        pw_main_loop_run(loop_);
    });

    return true;
}

void PipeMini::stop() {
    if (!running_) return;

    if (data_ && loop_) {
        pw_loop* loop = pw_main_loop_get_loop(loop_);
        pw_loop_invoke(loop, on_teardown_event, 0, nullptr, 0, true, data_);

        std::unique_lock<std::mutex> lock(data_->mtx);
        data_->cv.wait(lock, [this]() { return data_->teardown_done; });

        if (data_->loop_thread.joinable()) {
            data_->loop_thread.join();
        }

        delete data_;
        data_ = nullptr;
    }

    pw_main_loop_destroy(loop_);
    loop_ = nullptr;

    callback_ = nullptr;
    running_ = false;
}

void PipeMini::on_process(void* userdata) {
    auto* self = static_cast<PipeMini*>(userdata);
    if (!self || !self->callback_) return;

    pw_buffer* b = pw_stream_dequeue_buffer(self->stream_);
    if (!b) return;

    spa_buffer* buf = b->buffer;
    int16_t* dst = static_cast<int16_t*>(buf->datas[0].data);
    if (!dst) {
        pw_stream_queue_buffer(self->stream_, b);
        return;
    }

    uint32_t n_frames = buf->datas[0].maxsize / (DEFAULT_CHANNELS * sizeof(int16_t));
    if (b->requested) {
        n_frames = SPA_MIN(b->requested, n_frames);
    }

    for (uint32_t i = 0; i < n_frames; ++i) {
        int16_t left = 0;
        int16_t right = 0;
        self->callback_(&left, &right);
        dst[i * 2 + 0] = left;
        dst[i * 2 + 1] = right;
    }

    buf->datas[0].chunk->offset = 0;
    buf->datas[0].chunk->stride = DEFAULT_CHANNELS * sizeof(int16_t);
    buf->datas[0].chunk->size = n_frames * DEFAULT_CHANNELS * sizeof(int16_t);

    pw_stream_queue_buffer(self->stream_, b);
}