async_container.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "task.h"
#include "generator.h"
#include "core/container.h"
#include "core/container/error_codes.h"
 
#include <algorithm>
#include <atomic>
#include <fstream>
#include <functional>
#include <memory>
#include <optional>
#include <span>
#include <string>
#include <thread>
#include <vector>
 
namespace kcenon::container::async
{
    using progress_callback = std::function<void(size_t bytes_processed, size_t total_bytes)>;
 
    namespace detail
    {
        template<typename T>
        struct async_state
        {
            std::function<T()> work_;
            std::optional<T> result_;
            std::exception_ptr exception_;
            std::atomic<bool> ready_{false};
 
            explicit async_state(std::function<T()> work)
                : work_(std::move(work)) {}
 
            // Non-copyable, non-movable
            async_state(const async_state&) = delete;
            async_state& operator=(const async_state&) = delete;
            async_state(async_state&&) = delete;
            async_state& operator=(async_state&&) = delete;
        };
 
        template<typename T>
        struct async_awaitable
        {
            std::shared_ptr<async_state<T>> state_;
 
            explicit async_awaitable(std::function<T()> work)
                : state_(std::make_shared<async_state<T>>(std::move(work))) {}
 
            async_awaitable(async_awaitable&&) noexcept = default;
            async_awaitable& operator=(async_awaitable&&) noexcept = default;
 
            async_awaitable(const async_awaitable&) = delete;
            async_awaitable& operator=(const async_awaitable&) = delete;
 
            [[nodiscard]] bool await_ready() const noexcept
            {
                return false;
            }
 
            void await_suspend(std::coroutine_handle<> handle)
            {
                // Copy shared_ptr for thread to ensure state lifetime
                auto state = state_;
                std::thread([state, handle]() mutable {
                    try {
                        state->result_.emplace(state->work_());
                    } catch (...) {
                        state->exception_ = std::current_exception();
                    }
                    // Release barrier ensures all writes above are visible
                    // to the thread that reads with acquire semantics
                    state->ready_.store(true, std::memory_order_release);
                    handle.resume();
                }).detach();
            }
 
            T await_resume()
            {
                // Acquire barrier synchronizes with the release store above,
                // ensuring all writes in the worker thread are visible here
                while (!state_->ready_.load(std::memory_order_acquire)) {
                    // Spin until ready - in practice, handle.resume()
                    // ensures we only get here after ready_ is set
                }
                if (state_->exception_) {
                    std::rethrow_exception(state_->exception_);
                }
                return std::move(*state_->result_);
            }
        };
 
        template<typename F>
        auto make_async_awaitable(F&& func) -> async_awaitable<std::invoke_result_t<F>>
        {
            using result_type = std::invoke_result_t<F>;
            return async_awaitable<result_type>(std::forward<F>(func));
        }
 
    } // namespace detail
 
    class async_container
    {
    public:
        explicit async_container(std::shared_ptr<value_container> container)
            : container_(std::move(container))
        {
        }
 
        async_container()
            : container_(std::make_shared<value_container>())
        {
        }
 
        async_container(async_container&& other) noexcept = default;
 
        async_container& operator=(async_container&& other) noexcept = default;
 
        ~async_container() = default;
 
        // Non-copyable
        async_container(const async_container&) = delete;
        async_container& operator=(const async_container&) = delete;
 
        [[nodiscard]] std::shared_ptr<value_container> get_container() const noexcept
        {
            return container_;
        }
 
        void set_container(std::shared_ptr<value_container> container) noexcept
        {
            container_ = std::move(container);
        }
 
        // =======================================================================
        // Async Serialization APIs
        // =======================================================================
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] task<kcenon::common::Result<std::vector<uint8_t>>> serialize_async() const;
#else
        [[nodiscard]] task<std::vector<uint8_t>> serialize_async() const;
#endif
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] task<kcenon::common::Result<std::string>> serialize_string_async() const;
#else
        [[nodiscard]] task<std::string> serialize_string_async() const;
#endif
 
        // =======================================================================
        // Async Deserialization APIs
        // =======================================================================
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] static task<kcenon::common::Result<std::shared_ptr<value_container>>>
            deserialize_async(std::span<const uint8_t> data);
#else
        [[nodiscard]] static task<std::shared_ptr<value_container>>
            deserialize_async(std::span<const uint8_t> data);
#endif
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] static task<kcenon::common::Result<std::shared_ptr<value_container>>>
            deserialize_string_async(std::string_view data);
#else
        [[nodiscard]] static task<std::shared_ptr<value_container>>
            deserialize_string_async(std::string_view data);
#endif
 
        // =======================================================================
        // Convenience Methods (forwarding to underlying container)
        // =======================================================================
 
        template<typename T>
        async_container& set(std::string_view key, T&& value)
        {
            container_->set(key, std::forward<T>(value));
            return *this;
        }
 
        template<typename T>
        [[nodiscard]] std::optional<T> get(std::string_view key) const
        {
            auto val = container_->get(key);
            if (!val) {
                return std::nullopt;
            }
            if (auto* ptr = std::get_if<T>(&val->data)) {
                return *ptr;
            }
            return std::nullopt;
        }
 
        [[nodiscard]] bool contains(std::string_view key) const noexcept
        {
            return container_->contains(key);
        }
 
        // =======================================================================
        // Async File I/O APIs (Issue #267 - Phase 3)
        // =======================================================================
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] task<kcenon::common::VoidResult> load_async(
            std::string_view path,
            progress_callback callback = nullptr);
#else
        [[nodiscard]] task<bool> load_async(
            std::string_view path,
            progress_callback callback = nullptr);
#endif
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] task<kcenon::common::VoidResult> save_async(
            std::string_view path,
            progress_callback callback = nullptr);
#else
        [[nodiscard]] task<bool> save_async(
            std::string_view path,
            progress_callback callback = nullptr);
#endif
 
        // =======================================================================
        // Streaming APIs (Issue #268 - Phase 4)
        // =======================================================================
 
        [[nodiscard]] generator<std::vector<uint8_t>> serialize_chunked(
            size_t chunk_size = 64 * 1024) const;
 
#if CONTAINER_HAS_COMMON_RESULT
        [[nodiscard]] static task<kcenon::common::Result<std::shared_ptr<value_container>>>
            deserialize_streaming(
                std::span<const uint8_t> data,
                bool is_final = true);
#else
        [[nodiscard]] static task<std::shared_ptr<value_container>>
            deserialize_streaming(
                std::span<const uint8_t> data,
                bool is_final = true);
#endif
 
    private:
        std::shared_ptr<value_container> container_;
    };
 
    // ==========================================================================
    // Async File I/O Utility Functions (Issue #267 - Phase 3)
    // ==========================================================================
 
#if CONTAINER_HAS_COMMON_RESULT
    [[nodiscard]] task<kcenon::common::Result<std::vector<uint8_t>>> read_file_async(
        std::string_view path,
        progress_callback callback = nullptr);
#else
    [[nodiscard]] task<std::vector<uint8_t>> read_file_async(
        std::string_view path,
        progress_callback callback = nullptr);
#endif
 
#if CONTAINER_HAS_COMMON_RESULT
    [[nodiscard]] task<kcenon::common::VoidResult> write_file_async(
        std::string_view path,
        std::span<const uint8_t> data,
        progress_callback callback = nullptr);
#else
    [[nodiscard]] task<bool> write_file_async(
        std::string_view path,
        std::span<const uint8_t> data,
        progress_callback callback = nullptr);
#endif
 
    // ==========================================================================
    // Implementation of async methods (separate to avoid compiler issues)
    // ==========================================================================
 
#if CONTAINER_HAS_COMMON_RESULT
    inline task<kcenon::common::Result<std::vector<uint8_t>>>
        async_container::serialize_async() const
    {
        auto container = container_;
        auto result = co_await detail::make_async_awaitable([container]() {
            return container->serialize(value_container::serialization_format::binary);
        });
        co_return result;
    }
 
    inline task<kcenon::common::Result<std::string>>
        async_container::serialize_string_async() const
    {
        auto container = container_;
        auto result = co_await detail::make_async_awaitable([container]() {
            return container->serialize_string(value_container::serialization_format::binary);
        });
        co_return result;
    }
 
    inline task<kcenon::common::Result<std::shared_ptr<value_container>>>
        async_container::deserialize_async(std::span<const uint8_t> data)
    {
        std::vector<uint8_t> data_copy(data.begin(), data.end());
        auto result = co_await detail::make_async_awaitable(
            [data_copy = std::move(data_copy)]() {
                auto container = std::make_shared<value_container>();
                auto deser_result = container->deserialize_result(data_copy, false);
                if (!deser_result.is_ok()) {
                    return kcenon::common::Result<std::shared_ptr<value_container>>(
                        deser_result.error());
                }
                return kcenon::common::ok(container);
            });
        co_return result;
    }
 
    inline task<kcenon::common::Result<std::shared_ptr<value_container>>>
        async_container::deserialize_string_async(std::string_view data)
    {
        std::string data_copy(data);
        auto result = co_await detail::make_async_awaitable(
            [data_copy = std::move(data_copy)]() {
                auto container = std::make_shared<value_container>();
                auto deser_result = container->deserialize_result(data_copy, false);
                if (!deser_result.is_ok()) {
                    return kcenon::common::Result<std::shared_ptr<value_container>>(
                        deser_result.error());
                }
                return kcenon::common::ok(container);
            });
        co_return result;
    }
 
    inline task<kcenon::common::VoidResult>
        async_container::load_async(std::string_view path, progress_callback callback)
    {
        using namespace kcenon::container;
        auto container = container_;
        std::string path_str(path);
        auto result = co_await detail::make_async_awaitable(
            [container, path_str, callback]() -> kcenon::common::VoidResult {
                std::ifstream file(path_str, std::ios::binary | std::ios::ate);
                if (!file) {
                    return kcenon::common::VoidResult(
                        kcenon::common::error_info{
                            kcenon::container::error_codes::file_not_found,
                            kcenon::container::error_codes::make_message(
                                kcenon::container::error_codes::file_not_found, path_str),
                            "container_system"});
                }
 
                auto size = file.tellg();
                if (size < 0) {
                    return kcenon::common::VoidResult(
                        kcenon::common::error_info{
                            kcenon::container::error_codes::file_read_error,
                            kcenon::container::error_codes::make_message(
                                kcenon::container::error_codes::file_read_error, path_str),
                            "container_system"});
                }
                file.seekg(0, std::ios::beg);
 
                std::vector<uint8_t> buffer(static_cast<size_t>(size));
                const size_t chunk_size = 64 * 1024;
                size_t bytes_read = 0;
 
                while (bytes_read < static_cast<size_t>(size)) {
                    size_t to_read = std::min(chunk_size,
                        static_cast<size_t>(size) - bytes_read);
                    file.read(reinterpret_cast<char*>(buffer.data() + bytes_read),
                        static_cast<std::streamsize>(to_read));
                    if (!file) {
                        return kcenon::common::VoidResult(
                            kcenon::common::error_info{
                                kcenon::container::error_codes::file_read_error,
                                kcenon::container::error_codes::make_message(
                                    kcenon::container::error_codes::file_read_error, path_str),
                                "container_system"});
                    }
                    bytes_read += to_read;
                    if (callback) {
                        callback(bytes_read, static_cast<size_t>(size));
                    }
                }
 
                auto deser_result = container->deserialize_result(buffer, false);
                if (!deser_result.is_ok()) {
                    return deser_result;
                }
                return kcenon::common::ok();
            });
        co_return result;
    }
 
    inline task<kcenon::common::VoidResult>
        async_container::save_async(std::string_view path, progress_callback callback)
    {
        auto container = container_;
        std::string path_str(path);
        auto result = co_await detail::make_async_awaitable(
            [container, path_str, callback]() -> kcenon::common::VoidResult {
                auto data_result = container->serialize(value_container::serialization_format::binary);
                if (!data_result.is_ok()) {
                    return kcenon::common::VoidResult(data_result.error());
                }
                const auto& data = data_result.value();
 
                std::ofstream file(path_str, std::ios::binary);
                if (!file) {
                    return kcenon::common::VoidResult(
                        kcenon::common::error_info{
                            kcenon::container::error_codes::file_write_error,
                            kcenon::container::error_codes::make_message(
                                kcenon::container::error_codes::file_write_error, path_str),
                            "container_system"});
                }
 
                const size_t chunk_size = 64 * 1024;
                size_t bytes_written = 0;
                const size_t total_size = data.size();
 
                while (bytes_written < total_size) {
                    size_t to_write = std::min(chunk_size, total_size - bytes_written);
                    file.write(reinterpret_cast<const char*>(data.data() + bytes_written),
                        static_cast<std::streamsize>(to_write));
                    if (!file) {
                        return kcenon::common::VoidResult(
                            kcenon::common::error_info{
                                kcenon::container::error_codes::file_write_error,
                                kcenon::container::error_codes::make_message(
                                    kcenon::container::error_codes::file_write_error, path_str),
                                "container_system"});
                    }
                    bytes_written += to_write;
                    if (callback) {
                        callback(bytes_written, total_size);
                    }
                }
 
                return kcenon::common::ok();
            });
        co_return result;
    }
 
    inline task<kcenon::common::Result<std::vector<uint8_t>>>
        read_file_async(std::string_view path, progress_callback callback)
    {
        std::string path_str(path);
        auto result = co_await detail::make_async_awaitable(
            [path_str, callback]() -> kcenon::common::Result<std::vector<uint8_t>> {
                std::ifstream file(path_str, std::ios::binary | std::ios::ate);
                if (!file) {
                    return kcenon::common::Result<std::vector<uint8_t>>(
                        kcenon::common::error_info{
                            kcenon::container::error_codes::file_not_found,
                            kcenon::container::error_codes::make_message(
                                kcenon::container::error_codes::file_not_found, path_str),
                            "container_system"});
                }
 
                auto size = file.tellg();
                if (size < 0) {
                    return kcenon::common::Result<std::vector<uint8_t>>(
                        kcenon::common::error_info{
                            kcenon::container::error_codes::file_read_error,
                            kcenon::container::error_codes::make_message(
                                kcenon::container::error_codes::file_read_error, path_str),
                            "container_system"});
                }
                file.seekg(0, std::ios::beg);
 
                std::vector<uint8_t> buffer(static_cast<size_t>(size));
                const size_t chunk_size = 64 * 1024;
                size_t bytes_read = 0;
 
                while (bytes_read < static_cast<size_t>(size)) {
                    size_t to_read = std::min(chunk_size,
                        static_cast<size_t>(size) - bytes_read);
                    file.read(reinterpret_cast<char*>(buffer.data() + bytes_read),
                        static_cast<std::streamsize>(to_read));
                    if (!file) {
                        return kcenon::common::Result<std::vector<uint8_t>>(
                            kcenon::common::error_info{
                                kcenon::container::error_codes::file_read_error,
                                kcenon::container::error_codes::make_message(
                                    kcenon::container::error_codes::file_read_error, path_str),
                                "container_system"});
                    }
                    bytes_read += to_read;
                    if (callback) {
                        callback(bytes_read, static_cast<size_t>(size));
                    }
                }
 
                return kcenon::common::ok(std::move(buffer));
            });
        co_return result;
    }
 
    inline task<kcenon::common::VoidResult>
        write_file_async(std::string_view path, std::span<const uint8_t> data,
            progress_callback callback)
    {
        std::string path_str(path);
        std::vector<uint8_t> data_copy(data.begin(), data.end());
        auto result = co_await detail::make_async_awaitable(
            [path_str, data_copy = std::move(data_copy), callback]()
                -> kcenon::common::VoidResult {
                std::ofstream file(path_str, std::ios::binary);
                if (!file) {
                    return kcenon::common::VoidResult(
                        kcenon::common::error_info{
                            kcenon::container::error_codes::file_write_error,
                            kcenon::container::error_codes::make_message(
                                kcenon::container::error_codes::file_write_error, path_str),
                            "container_system"});
                }
 
                const size_t chunk_size = 64 * 1024;
                size_t bytes_written = 0;
                const size_t total_size = data_copy.size();
 
                while (bytes_written < total_size) {
                    size_t to_write = std::min(chunk_size, total_size - bytes_written);
                    file.write(reinterpret_cast<const char*>(data_copy.data() + bytes_written),
                        static_cast<std::streamsize>(to_write));
                    if (!file) {
                        return kcenon::common::VoidResult(
                            kcenon::common::error_info{
                                kcenon::container::error_codes::file_write_error,
                                kcenon::container::error_codes::make_message(
                                    kcenon::container::error_codes::file_write_error, path_str),
                                "container_system"});
                    }
                    bytes_written += to_write;
                    if (callback) {
                        callback(bytes_written, total_size);
                    }
                }
 
                return kcenon::common::ok();
            });
        co_return result;
    }
 
    // ==========================================================================
    // Streaming Implementation (Issue #268 - Phase 4)
    // ==========================================================================
 
    inline generator<std::vector<uint8_t>>
        async_container::serialize_chunked(size_t chunk_size) const
    {
        auto data_result = container_->serialize(value_container::serialization_format::binary);
        if (!data_result.is_ok()) {
            co_return;
        }
 
        const auto& full_data = data_result.value();
        const size_t total_size = full_data.size();
        size_t offset = 0;
 
        while (offset < total_size) {
            size_t current_chunk_size = std::min(chunk_size, total_size - offset);
            std::vector<uint8_t> chunk(
                full_data.begin() + static_cast<std::ptrdiff_t>(offset),
                full_data.begin() + static_cast<std::ptrdiff_t>(offset + current_chunk_size));
            offset += current_chunk_size;
            co_yield chunk;
        }
    }
 
    inline task<kcenon::common::Result<std::shared_ptr<value_container>>>
        async_container::deserialize_streaming(
            std::span<const uint8_t> data,
            bool is_final)
    {
        // For streaming deserialization, we accumulate data until we have a complete
        // container. In this simplified implementation, we require is_final to be true
        // to actually deserialize. For true streaming, a stateful deserializer would
        // be needed.
        if (!is_final) {
            // Return an error indicating more data is needed
            co_return kcenon::common::Result<std::shared_ptr<value_container>>(
                kcenon::common::error_info{
                    kcenon::container::error_codes::deserialization_failed,
                    kcenon::container::error_codes::make_message(
                        kcenon::container::error_codes::deserialization_failed,
                        "streaming requires complete data (is_final=true)"),
                    "container_system"});
        }
 
        std::vector<uint8_t> data_copy(data.begin(), data.end());
        auto result = co_await detail::make_async_awaitable(
            [data_copy = std::move(data_copy)]() {
                auto container = std::make_shared<value_container>();
                auto deser_result = container->deserialize_result(data_copy, false);
                if (!deser_result.is_ok()) {
                    return kcenon::common::Result<std::shared_ptr<value_container>>(
                        deser_result.error());
                }
                return kcenon::common::ok(container);
            });
        co_return result;
    }
 
#else
 
    inline task<std::vector<uint8_t>>
        async_container::serialize_async() const
    {
        auto container = container_;
        auto result = co_await detail::make_async_awaitable([container]() {
            auto serialize_result = container->serialize(serialization_format::binary);
            return serialize_result.is_ok() ? serialize_result.value() : std::vector<uint8_t>{};
        });
        co_return result;
    }
 
    inline task<std::string>
        async_container::serialize_string_async() const
    {
        auto container = container_;
        auto result = co_await detail::make_async_awaitable([container]() {
            auto serialize_result = container->serialize_string(serialization_format::binary);
            return serialize_result.is_ok() ? serialize_result.value() : std::string{};
        });
        co_return result;
    }
 
    inline task<std::shared_ptr<value_container>>
        async_container::deserialize_async(std::span<const uint8_t> data)
    {
        std::vector<uint8_t> data_copy(data.begin(), data.end());
        auto result = co_await detail::make_async_awaitable(
            [data_copy = std::move(data_copy)]() {
                auto container = std::make_shared<value_container>(data_copy, false);
                return container;
            });
        co_return result;
    }
 
    inline task<std::shared_ptr<value_container>>
        async_container::deserialize_string_async(std::string_view data)
    {
        std::string data_copy(data);
        auto result = co_await detail::make_async_awaitable(
            [data_copy = std::move(data_copy)]() {
                auto container = std::make_shared<value_container>(data_copy, false);
                return container;
            });
        co_return result;
    }
 
    inline task<bool>
        async_container::load_async(std::string_view path, progress_callback callback)
    {
        auto container = container_;
        std::string path_str(path);
        auto result = co_await detail::make_async_awaitable(
            [container, path_str, callback]() {
                std::ifstream file(path_str, std::ios::binary | std::ios::ate);
                if (!file) {
                    return false;
                }
 
                auto size = file.tellg();
                if (size < 0) {
                    return false;
                }
                file.seekg(0, std::ios::beg);
 
                std::vector<uint8_t> buffer(static_cast<size_t>(size));
                const size_t chunk_size = 64 * 1024;  // 64KB chunks
                size_t bytes_read = 0;
 
                while (bytes_read < static_cast<size_t>(size)) {
                    size_t to_read = std::min(chunk_size,
                        static_cast<size_t>(size) - bytes_read);
                    file.read(reinterpret_cast<char*>(buffer.data() + bytes_read),
                        static_cast<std::streamsize>(to_read));
                    if (!file) {
                        return false;
                    }
                    bytes_read += to_read;
                    if (callback) {
                        callback(bytes_read, static_cast<size_t>(size));
                    }
                }
 
                container->deserialize(buffer, false);
                return true;
            });
        co_return result;
    }
 
    inline task<bool>
        async_container::save_async(std::string_view path, progress_callback callback)
    {
        auto container = container_;
        std::string path_str(path);
        auto result = co_await detail::make_async_awaitable(
            [container, path_str, callback]() {
                auto serialize_result = container->serialize(serialization_format::binary);
                if (!serialize_result.is_ok()) {
                    return false;
                }
                auto data = serialize_result.value();
                if (data.empty()) {
                    return false;
                }
 
                std::ofstream file(path_str, std::ios::binary);
                if (!file) {
                    return false;
                }
 
                const size_t chunk_size = 64 * 1024;  // 64KB chunks
                size_t bytes_written = 0;
                const size_t total_size = data.size();
 
                while (bytes_written < total_size) {
                    size_t to_write = std::min(chunk_size, total_size - bytes_written);
                    file.write(reinterpret_cast<const char*>(data.data() + bytes_written),
                        static_cast<std::streamsize>(to_write));
                    if (!file) {
                        return false;
                    }
                    bytes_written += to_write;
                    if (callback) {
                        callback(bytes_written, total_size);
                    }
                }
 
                return true;
            });
        co_return result;
    }
 
    inline task<std::vector<uint8_t>>
        read_file_async(std::string_view path, progress_callback callback)
    {
        std::string path_str(path);
        auto result = co_await detail::make_async_awaitable(
            [path_str, callback]() {
                std::ifstream file(path_str, std::ios::binary | std::ios::ate);
                if (!file) {
                    return std::vector<uint8_t>{};
                }
 
                auto size = file.tellg();
                if (size < 0) {
                    return std::vector<uint8_t>{};
                }
                file.seekg(0, std::ios::beg);
 
                std::vector<uint8_t> buffer(static_cast<size_t>(size));
                const size_t chunk_size = 64 * 1024;  // 64KB chunks
                size_t bytes_read = 0;
 
                while (bytes_read < static_cast<size_t>(size)) {
                    size_t to_read = std::min(chunk_size,
                        static_cast<size_t>(size) - bytes_read);
                    file.read(reinterpret_cast<char*>(buffer.data() + bytes_read),
                        static_cast<std::streamsize>(to_read));
                    if (!file) {
                        return std::vector<uint8_t>{};
                    }
                    bytes_read += to_read;
                    if (callback) {
                        callback(bytes_read, static_cast<size_t>(size));
                    }
                }
 
                return buffer;
            });
        co_return result;
    }
 
    inline task<bool>
        write_file_async(std::string_view path, std::span<const uint8_t> data,
            progress_callback callback)
    {
        std::string path_str(path);
        std::vector<uint8_t> data_copy(data.begin(), data.end());
        auto result = co_await detail::make_async_awaitable(
            [path_str, data_copy = std::move(data_copy), callback]() {
                std::ofstream file(path_str, std::ios::binary);
                if (!file) {
                    return false;
                }
 
                const size_t chunk_size = 64 * 1024;  // 64KB chunks
                size_t bytes_written = 0;
                const size_t total_size = data_copy.size();
 
                while (bytes_written < total_size) {
                    size_t to_write = std::min(chunk_size, total_size - bytes_written);
                    file.write(reinterpret_cast<const char*>(data_copy.data() + bytes_written),
                        static_cast<std::streamsize>(to_write));
                    if (!file) {
                        return false;
                    }
                    bytes_written += to_write;
                    if (callback) {
                        callback(bytes_written, total_size);
                    }
                }
 
                return true;
            });
        co_return result;
    }
 
    // ==========================================================================
    // Streaming Implementation (Issue #268 - Phase 4) - Non-Result version
    // ==========================================================================
 
    inline generator<std::vector<uint8_t>>
        async_container::serialize_chunked(size_t chunk_size) const
    {
        auto serialize_result = container_->serialize(serialization_format::binary);
        if (!serialize_result.is_ok()) {
            co_return;
        }
        auto full_data = serialize_result.value();
        if (full_data.empty()) {
            co_return;
        }
 
        const size_t total_size = full_data.size();
        size_t offset = 0;
 
        while (offset < total_size) {
            size_t current_chunk_size = std::min(chunk_size, total_size - offset);
            std::vector<uint8_t> chunk(
                full_data.begin() + static_cast<std::ptrdiff_t>(offset),
                full_data.begin() + static_cast<std::ptrdiff_t>(offset + current_chunk_size));
            offset += current_chunk_size;
            co_yield chunk;
        }
    }
 
    inline task<std::shared_ptr<value_container>>
        async_container::deserialize_streaming(
            std::span<const uint8_t> data,
            bool is_final)
    {
        // For streaming deserialization, we accumulate data until we have a complete
        // container. In this simplified implementation, we require is_final to be true
        // to actually deserialize.
        if (!is_final) {
            co_return nullptr;
        }
 
        std::vector<uint8_t> data_copy(data.begin(), data.end());
        auto result = co_await detail::make_async_awaitable(
            [data_copy = std::move(data_copy)]() {
                auto container = std::make_shared<value_container>(data_copy, false);
                return container;
            });
        co_return result;
    }
 
#endif
 
} // namespace kcenon::container::async