reader.hpp

Go to the documentation of this file.

#ifndef LIBFILEZILLA_AIO_READER_HEADER
#define LIBFILEZILLA_AIO_READER_HEADER

#include "aio.hpp"
#include "../file.hpp"
#include "../thread_pool.hpp"

#include <list>

namespace fz {

class FZ_PUBLIC_SYMBOL reader_base : public aio_base, public aio_waitable, public event_handler
{
public:
    reader_base(reader_base const&) = delete;
    reader_base& operator=(reader_base const&) = delete;

    void close();

    virtual bool seekable() const { return false; }

    bool seek(uint64_t offset, uint64_t size = nosize);

    bool rewind();

    std::wstring const& name() const { return name_; }

    virtual uint64_t size() const { return size_; }

    virtual datetime mtime() const { return {}; }

    std::pair<aio_result, buffer_lease> get_buffer(event_handler & h);

    bool error() const;

protected:
    reader_base(std::wstring && name, aio_buffer_pool & pool, size_t max_buffers) noexcept
        : event_handler(pool.loop())
        , buffer_pool_(pool)
        , logger_(pool.logger())
        , name_(name)
        , max_buffers_(max_buffers ? max_buffers : 1)
    {}

    reader_base(std::wstring_view name, aio_buffer_pool & pool, size_t max_buffers) noexcept
        : event_handler(pool.loop())
        , buffer_pool_(pool)
        , logger_(pool.logger())
        , name_(name)
        , max_buffers_(max_buffers ? max_buffers : 1)
    {}

    virtual std::pair<aio_result, buffer_lease> do_get_buffer(scoped_lock & l) = 0;

    virtual bool do_seek(scoped_lock &) {
        return false;
    }

    virtual void do_close(scoped_lock &) {}

    virtual void operator()(event_base const&) override;
    void on_buffer_availability(aio_waitable const* w);
    virtual void do_on_buffer_availability(fz::scoped_lock & l, aio_waitable const* w) = 0;

    mutable mutex mtx_;
    aio_buffer_pool & buffer_pool_;
    logger_interface & logger_;

    std::wstring const name_;

    size_t const max_buffers_{};
    std::list<buffer_lease> buffers_;

    uint64_t size_{nosize};
    uint64_t max_size_{nosize};
    uint64_t start_offset_{nosize};
    uint64_t remaining_{nosize};

    bool get_buffer_called_{};
    bool error_{};
    bool eof_{};
    bool waiting_{};
};

class FZ_PUBLIC_SYMBOL reader_factory
{
public:
    explicit reader_factory(std::wstring const& name)
        : name_(name)
    {}
    explicit reader_factory(std::wstring && name)
        : name_(std::move(name))
    {}

    virtual ~reader_factory() noexcept = default;

    virtual std::unique_ptr<reader_factory> clone() const = 0;

    virtual std::unique_ptr<reader_base> open(aio_buffer_pool & pool, uint64_t offset = 0, uint64_t size = reader_base::nosize, size_t max_buffers = 0) = 0;

    virtual bool seekable() const { return false; }

    std::wstring name() const { return name_; }

    virtual uint64_t size() const { return reader_base::nosize; }
    virtual datetime mtime() const { return datetime(); }

    virtual size_t min_buffer_usage() const { return 1; }

    virtual bool multiple_buffer_usage() const { return false; }

    virtual size_t preferred_buffer_count() const { return 1; }

protected:
    reader_factory() = default;
    reader_factory(reader_factory const&) = default;

    std::wstring const name_;
};

class FZ_PUBLIC_SYMBOL reader_factory_holder final
{
public:
    reader_factory_holder() = default;
    reader_factory_holder(std::unique_ptr<reader_factory> && factory);
    reader_factory_holder(std::unique_ptr<reader_factory> const& factory);
    reader_factory_holder(reader_factory const& factory);

    reader_factory_holder(reader_factory_holder const& op);
    reader_factory_holder& operator=(reader_factory_holder const& op);

    reader_factory_holder(reader_factory_holder && op) noexcept;
    reader_factory_holder& operator=(reader_factory_holder && op) noexcept;
    reader_factory_holder& operator=(std::unique_ptr<reader_factory> && factory);

    reader_factory const* operator->() const { return impl_.get(); }
    reader_factory* operator->() { return impl_.get(); }
    reader_factory const& operator*() const { return *impl_; }
    reader_factory & operator*() { return *impl_; }

    explicit operator bool() const { return impl_.operator bool(); }

    std::wstring name() const { return impl_ ? impl_->name() : std::wstring(); }
    datetime mtime() const { return impl_ ? impl_->mtime() : datetime(); }
    uint64_t size() const { return impl_ ? impl_->size() : aio_base::nosize; }

private:
    std::unique_ptr<reader_factory> impl_;
};

class thread_pool;

class FZ_PUBLIC_SYMBOL threaded_reader : public reader_base
{
public:
    using reader_base::reader_base;

protected:
    virtual std::pair<aio_result, buffer_lease> do_get_buffer(scoped_lock & l) override;

    void wakeup(scoped_lock & l) {
        cond_.signal(l);
    }

    condition cond_;
    async_task task_;

    bool quit_{};
};

class FZ_PUBLIC_SYMBOL file_reader final : public threaded_reader
{
public:
    file_reader(std::wstring && name, aio_buffer_pool & pool, file && f, thread_pool & tpool, uint64_t offset = 0, uint64_t size = nosize, size_t max_buffers = 4) noexcept;
    file_reader(std::wstring_view name, aio_buffer_pool & pool, file && f, thread_pool & tpool, uint64_t offset = 0, uint64_t size = nosize, size_t max_buffers = 4) noexcept;

    virtual ~file_reader() noexcept;

    virtual bool seekable() const override;

private:
    virtual void FZ_PRIVATE_SYMBOL do_close(scoped_lock & l) override;
    virtual bool FZ_PRIVATE_SYMBOL do_seek(scoped_lock & l) override;

    virtual void FZ_PRIVATE_SYMBOL do_on_buffer_availability(scoped_lock & l, aio_waitable const* w) override;

    void FZ_PRIVATE_SYMBOL entry();

    file file_;
    thread_pool & thread_pool_;
};

class FZ_PUBLIC_SYMBOL file_reader_factory final : public reader_factory
{
public:
    file_reader_factory(std::wstring const& file, thread_pool & tpool);

    virtual std::unique_ptr<reader_base> open(aio_buffer_pool & pool, uint64_t offset = 0, uint64_t size = reader_base::nosize, size_t max_buffers = 4) override;
    virtual std::unique_ptr<reader_factory> clone() const override;

    virtual bool seekable() const override { return true; }

    virtual uint64_t size() const override;
    virtual datetime mtime() const override;

    virtual bool multiple_buffer_usage() const override { return true; }

    virtual size_t preferred_buffer_count() const override { return 4; }
private:
    thread_pool & thread_pool_;
};

class FZ_PUBLIC_SYMBOL view_reader final : public reader_base
{
public:
    view_reader(std::wstring && name, aio_buffer_pool & pool, std::string_view data) noexcept;

    virtual ~view_reader() noexcept;

    virtual bool seekable() const override { return true; }

private:
    virtual std::pair<aio_result, buffer_lease> FZ_PRIVATE_SYMBOL do_get_buffer(scoped_lock & l) override;
    virtual void FZ_PRIVATE_SYMBOL do_close(scoped_lock & l) override;
    virtual bool FZ_PRIVATE_SYMBOL do_seek(scoped_lock & l) override;

    virtual void FZ_PRIVATE_SYMBOL do_on_buffer_availability(scoped_lock & l, aio_waitable const* w) override;

    std::string_view const view_;
};

class FZ_PUBLIC_SYMBOL view_reader_factory final : public reader_factory
{
public:
    view_reader_factory(std::wstring && name, std::string_view const& view)
        : reader_factory(std::move(name))
        , view_(view)
    {}
    view_reader_factory(std::wstring const& name, std::string_view const& view)
        : reader_factory(name)
        , view_(view)
    {}

    virtual std::unique_ptr<reader_base> open(aio_buffer_pool & pool, uint64_t offset = 0, uint64_t size = reader_base::nosize, size_t max_buffers = 1) override;
    virtual std::unique_ptr<reader_factory> clone() const override;

    virtual bool seekable() const override { return true; }

    virtual uint64_t size() const override { return view_.size(); }

private:
    std::string_view const view_;
};

class FZ_PUBLIC_SYMBOL string_reader final : public reader_base
{
public:
    string_reader(std::wstring && name, aio_buffer_pool & pool, std::string const& data) noexcept;
    string_reader(std::wstring && name, aio_buffer_pool & pool, std::string && data) noexcept;

    virtual ~string_reader() noexcept;

    virtual bool seekable() const override { return true; }

private:
    virtual std::pair<aio_result, buffer_lease> FZ_PRIVATE_SYMBOL do_get_buffer(scoped_lock & l) override;
    virtual void FZ_PRIVATE_SYMBOL do_close(scoped_lock & l) override;
    virtual bool FZ_PRIVATE_SYMBOL do_seek(scoped_lock & l) override;

    virtual void FZ_PRIVATE_SYMBOL do_on_buffer_availability(scoped_lock & l, aio_waitable const* w) override;

    std::string const data_;
};

class FZ_PUBLIC_SYMBOL string_reader_factory final : public reader_factory
{
public:
    string_reader_factory(std::wstring const& name, std::string const& data)
        : reader_factory(name)
        , data_(data)
    {}
    string_reader_factory(std::wstring && name, std::string && data)
        : reader_factory(std::move(name))
        , data_(std::move(data))
    {}

    virtual std::unique_ptr<reader_base> open(aio_buffer_pool & pool, uint64_t offset = 0, uint64_t size = reader_base::nosize, size_t max_buffers = 1) override;
    virtual std::unique_ptr<reader_factory> clone() const override;

    virtual bool seekable() const override { return true; }

    virtual uint64_t size() const override { return data_.size(); }

private:
    std::string const data_;
};

}
#endif