◆ 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;
 
                const bool has_downstream_route = std::ranges::any_of(
                    m_operations,
                    [](const BufferOperation& o) {
                        return o.get_type() == BufferOperation::OpType::ROUTE;
                    });
 
                if (!has_downstream_route) {
                    for (auto& candidate : std::ranges::reverse_view(m_operations)) {
                        if (&candidate == &op)
                            continue;
                        if (candidate.get_type() != BufferOperation::OpType::CAPTURE
                            || !candidate.m_capture.get_buffer()) {
                            continue;
                        }
                        const auto buf = candidate.m_capture.get_buffer();
                        if (std::holds_alternative<std::vector<double>>(transformed)
                            && std::get<std::vector<double>>(transformed).size() == buf->get_data().size()) {
                            write_to_buffer(buf, 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());
    }
}
Referenced by execute_phased(), and execute_streaming().
Here is the call graph for this function:
Here is the caller graph for this function: