process_operation()

void MayaFlux::Kriya::BufferPipeline::process_operation ( BufferOperation &  op, uint64_t  cycle  )

private

Definition at line 511 of file BufferPipeline.cpp.

{
    try {
        switch (op.get_type()) {
        case BufferOperation::OpType::CAPTURE: {
            capture_operation(op, cycle);
 
            break;
        }
 
        case BufferOperation::OpType::TRANSFORM: {
            Kakshya::DataVariant input_data;
            if (m_operation_data.find(&op) != m_operation_data.end()) {
                input_data = m_operation_data[&op];
            } else {
                for (auto& it : m_operation_data) {
                    input_data = it.second;
                    break;
                }
            }
 
            if (op.m_transformer) {
                auto transformed = op.m_transformer(input_data, cycle);
                m_operation_data[&op] = transformed;
 
                for (auto& m_operation : std::ranges::reverse_view(m_operations)) {
                    if (&m_operation == &op)
                        continue;
                    if (m_operation.get_type() == BufferOperation::OpType::CAPTURE && m_operation.m_capture.get_buffer()) {
                        write_to_buffer(m_operation.m_capture.get_buffer(), transformed);
                        break;
                    }
                }
            }
            break;
        }
 
        case BufferOperation::OpType::ROUTE: {
            Kakshya::DataVariant data_to_route;
            if (m_operation_data.find(&op) != m_operation_data.end()) {
                data_to_route = m_operation_data[&op];
            } else {
                for (auto& it : m_operation_data) {
                    data_to_route = it.second;
                    break;
                }
            }
 
            if (op.m_target_buffer) {
                if (!m_buffer_manager) {
                    error<std::invalid_argument>(Journal::Component::Kriya,
                        Journal::Context::CoroutineScheduling,
                        std::source_location::current(),
                        "BufferPipeline has no BufferManager for ROUTE-to-buffer operation");
                }
 
                if (!op.m_attached_processor) {
                    auto writer = std::make_shared<Buffers::AudioWriteProcessor>();
                    m_buffer_manager->add_processor(writer, op.m_target_buffer,
                        Buffers::ProcessingToken::AUDIO_BACKEND);
                    op.m_attached_processor = writer;
                }
 
                std::static_pointer_cast<Buffers::AudioWriteProcessor>(op.m_attached_processor)
                    ->set_data(data_to_route);
 
            } else if (op.m_target_container) {
                write_to_container(op.m_target_container, data_to_route);
            }
            break;
        }
 
        case BufferOperation::OpType::LOAD: {
            auto loaded_data = read_from_container(op.m_source_container,
                op.m_start_frame,
                op.m_load_length);
 
            if (op.m_target_buffer) {
                write_to_buffer(op.m_target_buffer, loaded_data);
            }
 
            m_operation_data[&op] = loaded_data;
            break;
        }
 
        case BufferOperation::OpType::FUSE: {
            std::vector<Kakshya::DataVariant> fusion_inputs;
 
            for (auto& source_buffer : op.m_source_buffers) {
                bool should_process = op.m_capture.get_processing_control() == BufferCapture::ProcessingControl::ON_CAPTURE;
                auto buffer_data = extract_buffer_data(source_buffer, should_process);
                fusion_inputs.push_back(buffer_data);
            }
 
            for (auto& source_container : op.m_source_containers) {
                auto container_data = read_from_container(source_container, 0, 0);
                fusion_inputs.push_back(container_data);
            }
 
            if (op.m_fusion_function && !fusion_inputs.empty()) {
                auto fused_data = op.m_fusion_function(fusion_inputs, cycle);
 
                if (op.m_target_buffer) {
                    write_to_buffer(op.m_target_buffer, fused_data);
                } else if (op.m_target_container) {
                    write_to_container(op.m_target_container, fused_data);
                }
 
                m_operation_data[&op] = fused_data;
            }
            break;
        }
 
        case BufferOperation::OpType::DISPATCH: {
            Kakshya::DataVariant data_to_dispatch;
            if (m_operation_data.find(&op) != m_operation_data.end()) {
                data_to_dispatch = m_operation_data[&op];
            } else {
                for (auto& it : m_operation_data) {
                    data_to_dispatch = it.second;
                    break;
                }
            }
 
            if (op.m_dispatch_handler) {
                op.m_dispatch_handler(data_to_dispatch, cycle);
            }
            break;
        }
 
        case BufferOperation::OpType::MODIFY: {
            if (!m_buffer_manager) {
                error<std::invalid_argument>(Journal::Component::Kriya,
                    Journal::Context::CoroutineScheduling,
                    std::source_location::current(),
                    "BufferPipeline has no BufferManager for MODIFY operation");
            }
 
            if (!op.m_attached_processor) {
                op.m_attached_processor = m_buffer_manager->attach_quick_process(
                    op.m_buffer_modifier,
                    op.m_target_buffer, Buffers::ProcessingToken::AUDIO_BACKEND);
                if (m_max_cycles != 0 && op.is_streaming()) {
                    op.m_modify_cycle_count = m_max_cycles - cycle;
                }
            }
 
            if (op.m_modify_cycle_count > 0 && cycle >= op.m_modify_cycle_count - 1) {
                if (op.m_attached_processor) {
                    m_buffer_manager->remove_processor(
                        op.m_attached_processor,
                        op.m_target_buffer);
                    op.m_attached_processor = nullptr;
                }
            }
 
            break;
        }
 
        case BufferOperation::OpType::CONDITION:
            break;
 
        default:
            MF_ERROR(Journal::Component::Kriya,
                Journal::Context::CoroutineScheduling,
                "Unknown operation type in pipeline : {} : {}",
                Reflect::enum_to_string(op.get_type()), std::to_string(static_cast<int>(op.get_type())));
            break;
        }
    } catch (const std::exception& e) {
        error_rethrow(Journal::Component::Kriya,
            Journal::Context::CoroutineScheduling,
            std::source_location::current(),
            "Error processing operation in BufferPipeline: {}",
            e.what());
    }
}

References MayaFlux::Buffers::AUDIO_BACKEND, MayaFlux::Kriya::BufferOperation::CAPTURE, capture_operation(), MayaFlux::Kriya::BufferOperation::CONDITION, MayaFlux::Journal::CoroutineScheduling, MayaFlux::Kriya::BufferOperation::DISPATCH, MayaFlux::Reflect::enum_to_string(), extract_buffer_data(), MayaFlux::Kriya::BufferOperation::FUSE, MayaFlux::Kriya::BufferCapture::get_processing_control(), MayaFlux::Kriya::BufferOperation::get_type(), MayaFlux::Kriya::BufferOperation::is_streaming(), MayaFlux::Journal::Kriya, MayaFlux::Kriya::BufferOperation::LOAD, MayaFlux::Kriya::BufferOperation::m_attached_processor, m_buffer_manager, MayaFlux::Kriya::BufferOperation::m_buffer_modifier, MayaFlux::Kriya::BufferOperation::m_capture, MayaFlux::Kriya::BufferOperation::m_dispatch_handler, MayaFlux::Kriya::BufferOperation::m_fusion_function, MayaFlux::Kriya::BufferOperation::m_load_length, m_max_cycles, MayaFlux::Kriya::BufferOperation::m_modify_cycle_count, m_operation_data, m_operations, MayaFlux::Kriya::BufferOperation::m_source_buffers, MayaFlux::Kriya::BufferOperation::m_source_container, MayaFlux::Kriya::BufferOperation::m_source_containers, MayaFlux::Kriya::BufferOperation::m_start_frame, MayaFlux::Kriya::BufferOperation::m_target_buffer, MayaFlux::Kriya::BufferOperation::m_target_container, MayaFlux::Kriya::BufferOperation::m_transformer, MF_ERROR, MayaFlux::Kriya::BufferOperation::MODIFY, MayaFlux::Kriya::BufferCapture::ON_CAPTURE, read_from_container(), MayaFlux::Kriya::BufferOperation::ROUTE, MayaFlux::Kriya::BufferOperation::TRANSFORM, write_to_buffer(), and write_to_container().

Referenced by execute_phased(), and execute_streaming().

Here is the call graph for this function:

Here is the caller graph for this function: