async_stream.hpp

Go to the documentation of this file.

#pragma once
 
// AsyncStream — non-blocking wrapper around a synchronous Stream.
//
// AsyncStream owns a worker thread that pumps I/O on a Stream while the
// caller continues on its own thread. Read data and completion callbacks
// are delivered either on the worker thread (default) or dispatched into a
// Pulp EventLoop when one is supplied.
//
// The wrapper adds three things the raw Stream does not:
//   - asynchronous callbacks for read data, errors, and close;
//   - a bounded write queue with explicit backpressure (`write` returns
//     false when the pending queue exceeds a high-water mark);
//   - cancellation — destruction, `cancel()`, or `close()` stops all
//     in-flight work and fires pending completion callbacks with a
//     Cancelled error.
//
// This is layer 2 of the Stream hierarchy. Layer 1 (`Stream`) provides a
// synchronous interface; layer 3 adds network/HTTP-specific streams that
// are already `Stream` subclasses and therefore automatically work with
// `AsyncStream` without further adapter code.
 
#include <pulp/runtime/cancel.hpp>
#include <pulp/runtime/stream.hpp>
 
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <thread>
#include <vector>
 
namespace pulp::runtime {

using AsyncExecutor = std::function<void(std::function<void()>)>;

using AsyncWriteCallback = std::function<void(std::size_t bytes, StreamError error)>;

using AsyncDataCallback = std::function<void(const std::uint8_t* data, std::size_t size)>;

using AsyncErrorCallback = std::function<void(StreamError error)>;

using AsyncCloseCallback = std::function<void()>;
 
struct AsyncStreamOptions {
    std::size_t read_chunk = 4096;

    std::size_t write_high_water = 1 << 20;  // 1 MiB

    bool auto_read = true;

    AsyncExecutor executor;
};
 
class AsyncStream {
public:
    using Options = AsyncStreamOptions;
 
    explicit AsyncStream(std::unique_ptr<Stream> stream, Options options = {});
    ~AsyncStream();
 
    AsyncStream(const AsyncStream&) = delete;
    AsyncStream& operator=(const AsyncStream&) = delete;

    void on_data(AsyncDataCallback cb);
    void on_error(AsyncErrorCallback cb);
    void on_close(AsyncCloseCallback cb);
    void on_drain(AsyncCloseCallback cb);  

    void start();

    void stop();

    void cancel();

    [[nodiscard]] bool write_async(const std::uint8_t* data, std::size_t size,
                                   AsyncWriteCallback callback = {});

    std::size_t pending_write_bytes() const;

    CancelToken cancellation_token() const {
        return source_.token();
    }

    Stream* stream() { return stream_.get(); }
 
private:
    struct PendingWrite {
        std::vector<std::uint8_t> data;
        std::size_t written = 0;
        AsyncWriteCallback callback;
    };
 
    void write_loop();
    void read_loop();
    void dispatch(std::function<void()> task);
    void fire_close();
    void drain_queue_as_closed();  
 
    std::unique_ptr<Stream> stream_;
    Options options_;
    CancelSource source_;
 
    AsyncDataCallback on_data_;
    AsyncErrorCallback on_error_;
    AsyncCloseCallback on_close_;
    AsyncCloseCallback on_drain_;
 
    mutable std::mutex mutex_;
    std::condition_variable cv_;
    std::vector<PendingWrite> write_queue_;
    std::size_t pending_bytes_ = 0;
    bool running_ = false;
    bool close_fired_ = false;
    std::thread writer_thread_;
    std::thread reader_thread_;
};
 
}  // namespace pulp::runtime