io_stream.h

Go to the documentation of this file.

// Copyright (c) 2018, Smart Projects Holdings Ltd
// All rights reserved.
// See LICENSE file for license details.

#ifndef _UGCS_VSM_IO_STREAM_H_
#define _UGCS_VSM_IO_STREAM_H_

#include <ugcs/vsm/io_buffer.h>
#include <ugcs/vsm/request_temp_completion_context.h>
#include <ugcs/vsm/operation_waiter.h>
#include <ugcs/vsm/reference_guard.h>

#include <chrono>

namespace ugcs {
namespace vsm {

// Supported UDP packet size. Used as default with UDP sockets for reading.
// Read on UDP socket must reserve space for maximum payload it can possibly receive.
// Theoretical max is 64K which will always be fragmented.
// We use what is typically the largest unfragmented packet.
// This number is ether_v2 MTU - IP header - UDP header.
static constexpr int MIN_UDP_PAYLOAD_SIZE_TO_READ = 1500 - 60 - 8;

// TCP socket readers can use this as default max buffer size.
// Typically reads data worth one MTU.
// This number is ether_v2 MTU - IP header - TCP header (incl 20 byte options).
static constexpr int MAX_TCP_PAYLOAD_SIZE_TO_READ = 1500 - 60 - 40;

enum class Io_result {
    OK,
    TIMED_OUT,
    CANCELED,
    BAD_ADDRESS,
    CONNECTION_REFUSED,
    CLOSED,
    PERMISSION_DENIED,
    END_OF_FILE,
    LOCK_ERROR,
    OTHER_FAILURE
};

class Io_stream: public std::enable_shared_from_this<Io_stream> {
    DEFINE_COMMON_CLASS(Io_stream, Io_stream)

public:
    typedef Reference_guard<Io_stream::Ptr> Ref;

    typedef int64_t Offset;
    static const Offset OFFSET_NONE;
    static const Offset OFFSET_END;

    typedef Callback_proxy<void, Io_result> Write_handler;

    typedef Callback_proxy<void, Io_buffer::Ptr, Io_result> Read_handler;

    typedef Callback_proxy<void> Close_handler;

    enum class State {
        CLOSED,
        OPENING,
        OPENING_PASSIVE,
        OPENED
    };

    enum class Type {
        FILE,
        SERIAL,
        ANDROID_SERIAL,
        TCP,
        UDP,
        UDP_MULTICAST,
        CAN,
        UNDEFINED
    };

    virtual
    ~Io_stream()
    {}

    void
    Add_ref();

    void
    Release_ref();

    Operation_waiter
    Write(Io_buffer::Ptr buffer,
          Offset offset,
          Write_handler completion_handler = Make_dummy_callback<void, Io_result>(),
          Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create())
    {
        if (!!completion_handler != !!comp_ctx) {
            VSM_EXCEPTION(Invalid_param_exception, "Completion handler can not "
                    "exist without completion context and vice versa.");
        }
        return Write_impl(buffer, offset, completion_handler, comp_ctx);
    }

    Operation_waiter
    Write(Io_buffer::Ptr buffer,
          Write_handler completion_handler = Make_dummy_callback<void, Io_result>(),
          Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create())
    {
        if (!!completion_handler != !!comp_ctx) {
            VSM_EXCEPTION(Invalid_param_exception, "Completion handler can not "
                    "exist without completion context and vice versa.");
        }
        return Write_impl(buffer, OFFSET_NONE, completion_handler, comp_ctx);
    }

    Operation_waiter
    Read(size_t max_to_read, size_t min_to_read, Offset offset,
         Read_handler completion_handler = Make_dummy_callback<void, Io_buffer::Ptr, Io_result>(),
         Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create())
    {
        if (!!completion_handler != !!comp_ctx) {
            VSM_EXCEPTION(Invalid_param_exception, "Completion handler can not "
                    "exist without completion context and vice versa.");
        }
        if (max_to_read < min_to_read) {
            VSM_EXCEPTION(Invalid_param_exception, "max_to_read cannot be less than min_to_read");
        }
        return Read_impl(max_to_read, min_to_read, offset, completion_handler,
                         comp_ctx);
    }

    Operation_waiter
    Read(size_t max_to_read, size_t min_to_read = 1,
         Read_handler completion_handler = Make_dummy_callback<void, Io_buffer::Ptr, Io_result>(),
         Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create())
    {
        if (!!completion_handler != !!comp_ctx) {
            VSM_EXCEPTION(Invalid_param_exception, "Completion handler can not "
                    "exist without completion context and vice versa.");
        }
        if (min_to_read && max_to_read == 0) {
            // For UDP and TCP Let's try to read more than required as it can save us some syscalls later.
            // In UDP case we need to try read the maximum expected size. Otherwise we can get truncated data.
            switch (stream_type) {
            case Type::UDP:
            case Type::UDP_MULTICAST:
                max_to_read = MIN_UDP_PAYLOAD_SIZE_TO_READ;
                break;
            case Type::TCP:
                max_to_read = MAX_TCP_PAYLOAD_SIZE_TO_READ;
                break;
            default:
                max_to_read = min_to_read;
                break;
            }
        } else if (max_to_read < min_to_read) {
            VSM_EXCEPTION(Invalid_param_exception, "max_to_read cannot be less than min_to_read");
        }
        return Read_impl(max_to_read, min_to_read, OFFSET_NONE, completion_handler, comp_ctx);
    }

    Operation_waiter
    Close(Close_handler completion_handler = Make_dummy_callback<void>(),
          Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create())
    {
        if (!!completion_handler != !!comp_ctx) {
            VSM_EXCEPTION(Invalid_param_exception, "Completion handler can not "
                    "exist without completion context and vice versa.");
        }
        return Close_impl(completion_handler, comp_ctx);
    }

    Io_stream(const Io_stream&) = delete;

    Io_stream(Type type) :
        stream_type(type)
    {}

    State
    Get_state() const
    {
        return state;
    }

    bool
    Is_closed() const
    {
        return (state == State::CLOSED);
    }

    std::string
    Get_name() const;

    Type
    Get_type() const
    {
        return stream_type;
    }

    static const char*
    Io_result_as_char(const Io_result res);

protected:
    Type stream_type;

    State state = State::CLOSED;
    std::atomic_int ref_count = { 0 };

    virtual Operation_waiter
    Write_impl(Io_buffer::Ptr buffer, Offset offset,
               Write_handler completion_handler,
               Request_completion_context::Ptr comp_ctx) = 0;

    virtual Operation_waiter
    Read_impl(size_t max_to_read, size_t min_to_read, Offset offset,
              Read_handler completion_handler,
              Request_completion_context::Ptr comp_ctx) = 0;

    virtual Operation_waiter
    Close_impl(Close_handler completion_handler,
               Request_completion_context::Ptr comp_ctx) = 0;

    void
    Set_name(const std::string&);

private:
    static std::mutex name_mutex;

    std::string name = "[undefined]";
};

DEFINE_CALLBACK_BUILDER(Make_write_callback, (Io_result), (Io_result::OTHER_FAILURE))


DEFINE_CALLBACK_BUILDER(Make_read_callback, (Io_buffer::Ptr, Io_result),
                        (nullptr, Io_result::OTHER_FAILURE))

} /* namespace vsm */
} /* namespace ugcs */

#endif /* _UGCS_VSM_IO_STREAM_H_ */