stream.hpp

Go to the documentation of this file.

#pragma once
 
// Unified Stream interface for bytewise I/O.
//
// A `Stream` presents a common surface across files, memory buffers, pipes,
// sockets, and HTTP bodies. It is synchronous and byte-oriented; async
// behavior is layered on top via `AsyncStream`, which dispatches Stream
// operations on a background worker and delivers completions through the
// `EventLoop`.
//
// The interface is minimal on purpose:
//   - `read(buf, n)`  — non-blocking-style read; may return 0 with
//                        StreamError::WouldBlock when no data is available.
//   - `write(buf, n)` — returns bytes written; may be partial.
//   - `close()`        — idempotent; safe to call on a closed stream.
//   - `is_open()`      — whether the stream is currently usable.
//
// Streams never throw from `read`/`write`. Errors surface as
// `StreamError != Ok` in the returned `StreamResult`.
 
#include <cstddef>
#include <cstdint>
#include <memory>
#include <string>
#include <string_view>
#include <vector>
 
namespace pulp::runtime {

enum class StreamError : uint8_t {
    Ok,          
    Closed,      
    WouldBlock,  
    IoError,     
    Invalid,     
};

struct StreamResult {
    std::size_t bytes = 0;
    StreamError error = StreamError::Ok;
 
    constexpr bool ok() const { return error == StreamError::Ok; }
    constexpr bool would_block() const { return error == StreamError::WouldBlock; }
    constexpr bool closed() const { return error == StreamError::Closed; }
 
    static constexpr StreamResult make(std::size_t n) { return {n, StreamError::Ok}; }
    static constexpr StreamResult fail(StreamError e) { return {0, e}; }
};

class Stream {
public:
    virtual ~Stream() = default;
 
    Stream(const Stream&) = delete;
    Stream& operator=(const Stream&) = delete;

    virtual StreamResult read(std::uint8_t* buffer, std::size_t size) = 0;

    virtual StreamResult write(const std::uint8_t* buffer, std::size_t size) = 0;

    virtual void close() = 0;

    virtual bool is_open() const = 0;
 
protected:
    Stream() = default;
};

class FileStream : public Stream {
public:
    enum class Mode : uint8_t { Read, Write, Append, ReadWrite };
 
    FileStream() = default;
    explicit FileStream(std::string_view path, Mode mode = Mode::Read);
    ~FileStream() override;
 
    FileStream(FileStream&& other) noexcept;
    FileStream& operator=(FileStream&& other) noexcept;

    bool open(std::string_view path, Mode mode = Mode::Read);
 
    StreamResult read(std::uint8_t* buffer, std::size_t size) override;
    StreamResult write(const std::uint8_t* buffer, std::size_t size) override;
    void close() override;
    bool is_open() const override { return handle_ != nullptr; }

    bool flush();

    std::size_t position() const;
 
private:
    void* handle_ = nullptr;  // std::FILE*
};

class MemoryStream : public Stream {
public:
    MemoryStream() = default;

    explicit MemoryStream(std::vector<std::uint8_t> initial);
 
    StreamResult read(std::uint8_t* buffer, std::size_t size) override;
    StreamResult write(const std::uint8_t* buffer, std::size_t size) override;
    void close() override { open_ = false; }
    bool is_open() const override { return open_; }

    const std::vector<std::uint8_t>& buffer() const { return buffer_; }

    void rewind() { read_pos_ = 0; }

    void clear();
 
    std::size_t size() const { return buffer_.size(); }
    std::size_t read_position() const { return read_pos_; }
 
private:
    std::vector<std::uint8_t> buffer_;
    std::size_t read_pos_ = 0;
    bool open_ = true;
};
 
// Forward declaration — full definition requires named_pipe.hpp.
class NamedPipe;

class PipeStream : public Stream {
public:
    PipeStream() = default;
    explicit PipeStream(std::unique_ptr<NamedPipe> pipe);
    ~PipeStream() override;
 
    PipeStream(PipeStream&&) noexcept;
    PipeStream& operator=(PipeStream&&) noexcept;
 
    StreamResult read(std::uint8_t* buffer, std::size_t size) override;
    StreamResult write(const std::uint8_t* buffer, std::size_t size) override;
    void close() override;
    bool is_open() const override;
 
    NamedPipe* pipe() { return pipe_.get(); }
 
private:
    std::unique_ptr<NamedPipe> pipe_;
};
 
}  // namespace pulp::runtime