<commit_before>//
//  yas_audio_offline_output.cpp
//

#include "yas_audio_engine_offline_output.h"
#include "yas_audio_engine_node.h"
#include "yas_audio_time.h"
#include "yas_operation.h"
#include "yas_stl_utils.h"

using namespace yas;

#pragma mark - audio::engine::offline_output::impl

struct audio::engine::offline_output::impl : base::impl, manageable_offline_output::impl {
    operation_queue _queue = nullptr;
    audio::engine::node _node = {{.input_bus_count = 1, .output_bus_count = 0}};
    chaining::any_observer _reset_observer = nullptr;

    ~impl() = default;

    void prepare(offline_output const &output) {
        this->_reset_observer = this->_node.chain(node::method::will_reset)
                                    .perform([weak_output = to_weak(output)](auto const &) {
                                        if (auto output = weak_output.lock()) {
                                            output.impl_ptr<audio::engine::offline_output::impl>()->stop();
                                        }
                                    })
                                    .end();
    }

    audio::engine::offline_start_result_t start(offline_render_f &&render_handler,
                                                offline_completion_f &&completion_handler) override {
        if (this->_queue) {
            return offline_start_result_t(offline_start_error_t::already_running);
        } else if (auto connection = this->_node.input_connection(0)) {
            std::optional<uint8_t> key;
            if (completion_handler) {
                key = this->_core.push_completion_handler(std::move(completion_handler));
                if (!key) {
                    return offline_start_result_t(offline_start_error_t::prepare_failure);
                }
            }

            audio::pcm_buffer render_buffer(connection.format(), 1024);

            auto weak_output = to_weak(cast<offline_output>());
            auto operation_lambda = [weak_output, render_buffer, render_handler = std::move(render_handler),
                                     key](operation const &op) mutable {
                bool cancelled = false;
                uint32_t current_sample_time = 0;
                bool stop = false;

                while (!stop) {
                    audio::time when(current_sample_time, render_buffer.format().sample_rate());
                    auto output = weak_output.lock();
                    if (!output) {
                        cancelled = true;
                        break;
                    }

                    auto kernel = output.node().kernel();
                    if (!kernel) {
                        cancelled = true;
                        break;
                    }

                    auto connection_on_block = kernel.input_connection(0);
                    if (!connection_on_block) {
                        cancelled = true;
                        break;
                    }

                    auto format = connection_on_block.format();
                    if (format != render_buffer.format()) {
                        cancelled = true;
                        break;
                    }

                    render_buffer.reset();

                    if (auto src_node = connection_on_block.source_node()) {
                        src_node.render(
                            {.buffer = render_buffer, .bus_idx = connection_on_block.source_bus(), .when = when});
                    }

                    if (render_handler) {
                        render_handler({.buffer = render_buffer, .when = when, .out_stop = stop});
                    }

                    if (op.is_canceled()) {
                        cancelled = true;
                        break;
                    }

                    current_sample_time += 1024;
                }

                auto completion_lambda = [weak_output, cancelled, key]() {
                    if (auto output = weak_output.lock()) {
                        std::optional<offline_completion_f> completion_handler;
                        if (key) {
                            completion_handler = output.impl_ptr<impl>()->_core.pull_completion_handler(*key);
                        }

                        output.impl_ptr<impl>()->_queue = nullptr;

                        if (completion_handler) {
                            (*completion_handler)(cancelled);
                        }
                    }
                };

                dispatch_async(dispatch_get_main_queue(), completion_lambda);
            };

            operation operation{std::move(operation_lambda)};
            this->_queue = operation_queue{1};
            this->_queue.push_back(operation);
        } else {
            return offline_start_result_t(offline_start_error_t::connection_not_found);
        }
        return offline_start_result_t(nullptr);
    }

    void stop() override {
        auto completion_handlers = _core.pull_completion_handlers();

        if (auto &queue = this->_queue) {
            queue.cancel();
            queue.wait_until_all_operations_are_finished();
            this->_queue = nullptr;
        }

        for (auto &pair : completion_handlers) {
            auto &func = pair.second;
            if (func) {
                func(true);
            }
        }
    }

    bool is_running() {
        return this->_queue != nullptr;
    }

    audio::engine::node &node() {<commit_msg>fix operation cancel<commit_after>//
//  yas_audio_offline_output.cpp
//

#include "yas_audio_engine_offline_output.h"
#include "yas_audio_engine_node.h"
#include "yas_audio_time.h"
#include "yas_operation.h"
#include "yas_stl_utils.h"

using namespace yas;

#pragma mark - audio::engine::offline_output::impl

struct audio::engine::offline_output::impl : base::impl, manageable_offline_output::impl {
    operation_queue _queue = nullptr;
    audio::engine::node _node = {{.input_bus_count = 1, .output_bus_count = 0}};
    chaining::any_observer _reset_observer = nullptr;

    ~impl() = default;

    void prepare(offline_output const &output) {
        this->_reset_observer = this->_node.chain(node::method::will_reset)
                                    .perform([weak_output = to_weak(output)](auto const &) {
                                        if (auto output = weak_output.lock()) {
                                            output.impl_ptr<audio::engine::offline_output::impl>()->stop();
                                        }
                                    })
                                    .end();
    }

    audio::engine::offline_start_result_t start(offline_render_f &&render_handler,
                                                offline_completion_f &&completion_handler) override {
        if (this->_queue) {
            return offline_start_result_t(offline_start_error_t::already_running);
        } else if (auto connection = this->_node.input_connection(0)) {
            std::optional<uint8_t> key;
            if (completion_handler) {
                key = this->_core.push_completion_handler(std::move(completion_handler));
                if (!key) {
                    return offline_start_result_t(offline_start_error_t::prepare_failure);
                }
            }

            audio::pcm_buffer render_buffer(connection.format(), 1024);

            auto weak_output = to_weak(cast<offline_output>());
            auto operation_lambda = [weak_output, render_buffer, render_handler = std::move(render_handler),
                                     key](operation const &op) mutable {
                bool cancelled = false;
                uint32_t current_sample_time = 0;
                bool stop = false;

                while (!stop) {
                    audio::time when(current_sample_time, render_buffer.format().sample_rate());
                    auto output = weak_output.lock();
                    if (!output) {
                        cancelled = true;
                        break;
                    }

                    auto kernel = output.node().kernel();
                    if (!kernel) {
                        cancelled = true;
                        break;
                    }

                    auto connection_on_block = kernel.input_connection(0);
                    if (!connection_on_block) {
                        cancelled = true;
                        break;
                    }

                    auto format = connection_on_block.format();
                    if (format != render_buffer.format()) {
                        cancelled = true;
                        break;
                    }

                    render_buffer.reset();

                    if (auto src_node = connection_on_block.source_node()) {
                        src_node.render(
                            {.buffer = render_buffer, .bus_idx = connection_on_block.source_bus(), .when = when});
                    }

                    if (render_handler) {
                        render_handler({.buffer = render_buffer, .when = when, .out_stop = stop});
                    }

                    if (op.is_canceled()) {
                        cancelled = true;
                        break;
                    }

                    current_sample_time += 1024;
                }

                auto completion_lambda = [weak_output, cancelled, key]() {
                    if (auto output = weak_output.lock()) {
                        std::optional<offline_completion_f> completion_handler;
                        if (key) {
                            completion_handler = output.impl_ptr<impl>()->_core.pull_completion_handler(*key);
                        }

                        output.impl_ptr<impl>()->_queue = nullptr;

                        if (completion_handler) {
                            (*completion_handler)(cancelled);
                        }
                    }
                };

                dispatch_async(dispatch_get_main_queue(), completion_lambda);
            };

            operation operation{std::move(operation_lambda)};
            this->_queue = operation_queue{1};
            this->_queue.push_back(operation);
        } else {
            return offline_start_result_t(offline_start_error_t::connection_not_found);
        }
        return offline_start_result_t(nullptr);
    }

    void stop() override {
        auto completion_handlers = _core.pull_completion_handlers();

        if (auto &queue = this->_queue) {
            queue.cancel_all();
            queue.wait_until_all_operations_are_finished();
            this->_queue = nullptr;
        }

        for (auto &pair : completion_handlers) {
            auto &func = pair.second;
            if (func) {
                func(true);
            }
        }
    }

    bool is_running() {
        return this->_queue != nullptr;
    }

    audio::engine::node &node() {
{'fraction_non_alphanumeric': 0.09694760820045559, 'fraction_numerical': 0.0030979498861047837, 'mean_word_length': 1.3876441157276485, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7316062', 'n_tokens_mistral': 2837, 'n_tokens_neox': 2628, 'n_words': 460}