session.hpp

Go to the documentation of this file.

#pragma once
 
#include <roar/mechanics/cookie.hpp>
#include <functional>
#include <roar/error.hpp>
#include <roar/websocket/websocket_session.hpp>
#include <roar/response.hpp>
#include <roar/beast/forward.hpp>
#include <roar/detail/pimpl_special_functions.hpp>
#include <roar/literals/memory.hpp>
#include <roar/routing/proto_route.hpp>
#include <roar/standard_response_provider.hpp>
#include <roar/detail/promise_compat.hpp>
#include <roar/body/void_body.hpp>
#include <roar/detail/stream_type.hpp>
#include <roar/body/range_file_body.hpp>
 
#include <boost/beast/http/message.hpp>
#include <boost/beast/http/empty_body.hpp>
#include <boost/beast/http/parser.hpp>
#include <boost/beast/http/write.hpp>
#include <boost/beast/http/read.hpp>
#include <boost/beast/ssl/ssl_stream.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <roar/ssl/make_ssl_context.hpp>
#include <promise-cpp/promise.hpp>
 
#include <memory>
#include <optional>
#include <chrono>
#include <variant>
#include <stdexcept>
#include <sstream>
 
namespace Roar
{
    class Router;
    template <typename>
    class Request;
    class Session : public std::enable_shared_from_this<Session>
    {
      public:
        friend class Route;
        friend class Factory;
 
        constexpr static uint64_t defaultHeaderLimit{8_MiB};
        constexpr static uint64_t defaultBodyLimit{8_MiB};
        constexpr static std::chrono::seconds sessionTimeout{10};
 
        Session(
            boost::asio::basic_stream_socket<boost::asio::ip::tcp>&& socket,
            boost::beast::basic_flat_buffer<std::allocator<char>>&& buffer,
            std::optional<std::variant<SslServerContext, boost::asio::ssl::context>>& sslContext,
            bool isSecure,
            std::function<void(Error&&)> onError,
            std::weak_ptr<Router> router,
            std::shared_ptr<const StandardResponseProvider> standardResponseProvider);
        ROAR_PIMPL_SPECIAL_FUNCTIONS(Session);
 
        void close();
 
        template <typename BodyT>
        class SendIntermediate : public std::enable_shared_from_this<SendIntermediate<BodyT>>
        {
          public:
            template <typename... Forwards, typename RequestBodyT>
            SendIntermediate(Session& session, Request<RequestBodyT> const& req, Forwards&&... forwards)
                : session_(session.shared_from_this())
                , response_{session_->prepareResponse<BodyT>(req, std::forward<Forwards>(forwards)...)}
            {}
            SendIntermediate(Session& session, boost::beast::http::response<BodyT> res)
                : session_(session.shared_from_this())
                , response_{std::move(res)}
            {}
            SendIntermediate(Session& session, Response<BodyT>&& res)
                : session_(session.shared_from_this())
                , response_{std::move(res)}
            {}
            SendIntermediate(Session& session)
                : session_(session.shared_from_this())
                , response_{session_->prepareResponse<BodyT>()}
            {}
            SendIntermediate(SendIntermediate&&) = default;
            SendIntermediate(SendIntermediate const&) = delete;
            SendIntermediate& operator=(SendIntermediate&&) = default;
            SendIntermediate& operator=(SendIntermediate const&) = delete;
 
            SendIntermediate& status(boost::beast::http::status status)
            {
                response_.status(status);
                return *this;
            }
 
            auto& body()
            {
                return response_.body();
            }
 
            template <typename T>
            SendIntermediate& body(T&& toAssign)
            {
                response_.body(std::forward<T>(toAssign));
                preparePayload();
                return *this;
            }
 
            SendIntermediate& chunked(bool activate = true)
            {
                response_.chunked(activate);
                return *this;
            }
 
            template <typename T>
            SendIntermediate& contentType(T&& type)
            {
                response_.contentType(std::forward<T>(type));
                return *this;
            }
 
            template <typename T>
            SendIntermediate& setHeader(boost::beast::http::field field, T&& value)
            {
                response_.setHeader(field, std::forward<T>(value));
                return *this;
            }
            template <typename T>
            SendIntermediate& header(boost::beast::http::field field, T&& value)
            {
                return setHeader(field, std::forward<T>(value));
            }
 
            SendIntermediate& onWriteSome(std::function<bool(std::size_t)> onChunkFunc)
            {
                onChunk_ = onChunkFunc;
                return *this;
            }
 
            template <typename RequestBodyT2>
            SendIntermediate&
            enableCors(Request<RequestBodyT2> const& req, std::optional<CorsSettings> cors = std::nullopt)
            {
                response_.enableCors(req, std::move(cors));
                return *this;
            }
 
            SendIntermediate& preparePayload()
            {
                response_.preparePayload();
                return *this;
            }
 
            SendIntermediate& modifyResponse(std::function<void(Response<BodyT>&)> func)
            {
                func(response_);
                return *this;
            }
 
          private:
            using CommitReturnType =
                Detail::PromiseTypeBind<Detail::PromiseTypeBindThen<bool>, Detail::PromiseTypeBindFail<Error>>;
 
          public:
            template <typename T = BodyT, bool OverrideSfinae = false>
            std::enable_if_t<
                !(std::is_same_v<T, RangeFileBody> || std::is_same_v<T, boost::beast::http::empty_body>) ||
                    OverrideSfinae,
                CommitReturnType>
            commit()
            {
                if (overallTimeout_)
                {
                    session_->withStreamDo([this](auto& stream) {
                        boost::beast::get_lowest_layer(stream).expires_after(*overallTimeout_);
                    });
                }
                promise_ = std::make_unique<promise::Promise>(promise::newPromise());
                serializer_ = std::make_unique<boost::beast::http::serializer<false, BodyT>>(response_.response());
                writeChunk();
                return {*promise_};
            }
 
            template <typename T = BodyT>
            std::enable_if_t<std::is_same_v<T, boost::beast::http::empty_body>, CommitReturnType> commit()
            {
                if (overallTimeout_)
                {
                    session_->withStreamDo([this](auto& stream) {
                        boost::beast::get_lowest_layer(stream).expires_after(*overallTimeout_);
                    });
                }
                promise_ = std::make_unique<promise::Promise>(promise::newPromise());
                serializer_ = std::make_unique<boost::beast::http::serializer<false, BodyT>>(response_.response());
                writeHeader();
                return {*promise_};
            }
 
            template <typename T = BodyT>
            std::enable_if_t<std::is_same_v<T, RangeFileBody>, CommitReturnType> commit()
            {
                using namespace boost::beast::http;
                preparePayload();
                if (response_.body().isMultipart())
                    setHeader(field::content_type, "multipart/byteranges; boundary=" + response_.body().boundary());
                else
                {
                    const auto first = response_.body().firstRange().first;
                    const auto second = response_.body().firstRange().second;
                    const auto fileSize = response_.body().fileSize();
 
                    setHeader(
                        field::content_range,
                        std::string{"bytes "} + std::to_string(first) + "-" + std::to_string(second) + "/" +
                            std::to_string(fileSize));
                }
                status(status::partial_content);
                return commit<BodyT, true>();
            }
 
            SendIntermediate& rejectAuthorization(std::string const& wwwAuthenticate)
            {
                if (!wwwAuthenticate.empty())
                    response_.setHeader(boost::beast::http::field::www_authenticate, wwwAuthenticate);
                response_.status(boost::beast::http::status::unauthorized);
                return *this;
            }
 
            SendIntermediate& keepAlive(bool keepAlive = true)
            {
                response_.keepAlive(keepAlive);
                return *this;
            }
 
            SendIntermediate& setCookie(Cookie const& cookie)
            {
                response_.setCookie(cookie);
                return *this;
            }
 
            SendIntermediate& useFixedTimeout(std::chrono::milliseconds timeout)
            {
                overallTimeout_ = timeout;
                return *this;
            }
 
          private:
            void writeHeader()
            {
                session_->withStreamDo([this](auto& stream) {
                    if (!overallTimeout_)
                        boost::beast::get_lowest_layer(stream).expires_after(std::chrono::seconds(sessionTimeout));
 
                    boost::beast::http::async_write_header(
                        stream,
                        *serializer_,
                        [self = this->shared_from_this()](boost::beast::error_code ec, std::size_t) {
                            if (ec)
                                self->promise_->fail(ec);
                            else
                                self->promise_->resolve(false);
                        });
                });
            }
 
            void writeChunk()
            {
                session_->withStreamDo([this](auto& stream) {
                    if (!overallTimeout_)
                        boost::beast::get_lowest_layer(stream).expires_after(std::chrono::seconds(sessionTimeout));
 
                    boost::beast::http::async_write_some(
                        stream,
                        *serializer_,
                        [self = this->shared_from_this()](boost::beast::error_code ec, std::size_t bytesTransferred) {
                            if (!ec && !self->serializer_->is_done())
                                return self->writeChunk();
 
                            if (ec)
                            {
                                self->session_->close();
                                self->promise_->reject(Error{.error = ec, .additionalInfo = "Failed to send response"});
                                return;
                            }
 
                            if (self->onChunk_ && !self->onChunk_(bytesTransferred))
                                return;
 
                            try
                            {
                                self->promise_->resolve(self->session_->onWriteComplete(
                                    self->serializer_->get().need_eof(), ec, bytesTransferred));
                            }
                            catch (std::exception const& exc)
                            {
                                self->promise_->fail(
                                    Error{.error = exc.what(), .additionalInfo = "Failed to send response"});
                            }
                        });
                });
            }
 
          private:
            std::shared_ptr<Session> session_;
            Response<BodyT> response_;
            std::function<bool(std::size_t)> onChunk_;
            std::unique_ptr<promise::Promise> promise_;
            std::optional<std::chrono::milliseconds> overallTimeout_;
            std::unique_ptr<boost::beast::http::serializer<false, BodyT>> serializer_;
        };
 
        template <typename BodyT, typename OriginalBodyT, typename... Forwards>
        [[nodiscard]] std::shared_ptr<SendIntermediate<BodyT>>
        send(Request<OriginalBodyT> const& req, Forwards&&... forwards)
        {
            return std::shared_ptr<SendIntermediate<BodyT>>(
                new SendIntermediate<BodyT>{*this, req, std::forward<Forwards>(forwards)...});
        }
 
        template <typename BodyT>
        [[nodiscard]] std::shared_ptr<SendIntermediate<BodyT>> send(boost::beast::http::response<BodyT>&& res)
        {
            return std::shared_ptr<SendIntermediate<BodyT>>(new SendIntermediate<BodyT>{*this, std::move(res)});
        }
 
        template <typename BodyT>
        [[nodiscard]] std::shared_ptr<SendIntermediate<BodyT>> send(Response<BodyT>&& res)
        {
            return std::shared_ptr<SendIntermediate<BodyT>>(new SendIntermediate<BodyT>{*this, std::move(res)});
        }
 
        template <typename BodyT>
        [[nodiscard]] std::shared_ptr<SendIntermediate<BodyT>> sendWithAllAcceptedCors()
        {
            return std::shared_ptr<SendIntermediate<BodyT>>(new SendIntermediate<BodyT>{*this});
        }
 
        Detail::PromiseTypeBind<Detail::PromiseTypeBindThen<std::size_t>, Detail::PromiseTypeBindFail<Error>>
        send(std::string message, std::chrono::seconds timeout = std::chrono::seconds{10});
 
        template <typename BodyT>
        class ReadIntermediate : public std::enable_shared_from_this<ReadIntermediate<BodyT>>
        {
          private:
            friend Session;
 
            template <typename OriginalBodyT, typename... Forwards>
            ReadIntermediate(Session& session, Request<OriginalBodyT> req, Forwards&&... forwardArgs)
                : session_{session.shared_from_this()}
                , req_{[&session, &forwardArgs...]() -> decltype(req_) {
                    if constexpr (std::is_same_v<BodyT, boost::beast::http::empty_body>)
                        throw std::runtime_error("Attempting to read with empty_body type.");
                    else
                        return boost::beast::http::request_parser<BodyT>{
                            std::move(*session.parser()), std::forward<Forwards>(forwardArgs)...};
                    session.parser().reset();
                }()}
                , originalExtensions_{std::move(req).ejectExtensions()}
                , onChunk_{}
                , promise_{}
                , overallTimeout_{std::nullopt}
            {
                req_.body_limit(defaultBodyLimit);
            }
 
          public:
            ReadIntermediate(ReadIntermediate&&) = default;
            ReadIntermediate& operator=(ReadIntermediate&&) = default;
 
            ReadIntermediate& bodyLimit(std::size_t limit)
            {
                req_.body_limit(limit);
                return *this;
            }
 
            ReadIntermediate& bodyLimit(boost::beast::string_view limit)
            {
                req_.body_limit(std::stoull(std::string{limit}));
                return *this;
            }
 
            ReadIntermediate& noBodyLimit()
            {
                req_.body_limit(boost::none);
                return *this;
            }
 
            ReadIntermediate& onReadSome(std::function<bool(boost::beast::flat_buffer&, std::size_t)> onChunkFunc)
            {
                onChunk_ = onChunkFunc;
                return *this;
            }
 
            Detail::PromiseTypeBind<
                Detail::PromiseTypeBindThen<
                    Detail::PromiseReferenceWrap<Session>,
                    Detail::PromiseReferenceWrap<Request<BodyT> const>>,
                Detail::PromiseTypeBindFail<Error const&>>
            commit()
            {
                if (overallTimeout_)
                {
                    session_->withStreamDo([this](auto& stream) {
                        boost::beast::get_lowest_layer(stream).expires_after(*overallTimeout_);
                    });
                }
                promise_ = std::make_unique<promise::Promise>(promise::newPromise());
                readChunk();
                return {*promise_};
            }
 
            ReadIntermediate& useFixedTimeout(std::chrono::milliseconds timeout)
            {
                overallTimeout_ = timeout;
                return *this;
            }
 
          private:
            void readChunk()
            {
                session_->withStreamDo([this](auto& stream) {
                    if (!overallTimeout_)
                        boost::beast::get_lowest_layer(stream).expires_after(std::chrono::seconds(sessionTimeout));
                    boost::beast::http::async_read_some(
                        stream,
                        session_->buffer(),
                        req_,
                        [self = this->shared_from_this()](boost::beast::error_code ec, std::size_t bytesReceived) {
                            if (!ec && !self->req_.is_done())
                                return self->readChunk();
 
                            if (ec)
                            {
                                self->session_->close();
                                self->promise_->reject(Error{.error = ec});
                            }
 
                            if (self->onChunk_ && !self->onChunk_(self->session_->buffer(), bytesReceived))
                                return;
 
                            auto request = Request<BodyT>(self->req_.release(), std::move(self->originalExtensions_));
                            try
                            {
                                self->promise_->template resolve(Detail::ref(*self->session_), Detail::cref(request));
                            }
                            catch (std::exception const& exc)
                            {
                                using namespace std::string_literals;
                                self->session_
                                    ->send(self->session_->standardResponseProvider().makeStandardResponse(
                                        *self->session_,
                                        boost::beast::http::status::internal_server_error,
                                        "An exception was thrown in the body read completion handler: "s + exc.what()))
                                    ->commit();
                            }
                        });
                });
            }
 
          private:
            std::shared_ptr<Session> session_;
            boost::beast::http::request_parser<BodyT> req_;
            Detail::RequestExtensions originalExtensions_;
            std::function<bool(boost::beast::flat_buffer&, std::size_t)> onChunk_;
            std::unique_ptr<promise::Promise> promise_;
            std::optional<std::chrono::milliseconds> overallTimeout_;
        };
 
        void readLimit(std::size_t bytesPerSecond);
 
        void writeLimit(std::size_t bytesPerSecond);
 
        template <typename BodyT, typename OriginalBodyT, typename... Forwards>
        [[nodiscard]] std::shared_ptr<ReadIntermediate<BodyT>>
        read(Request<OriginalBodyT> req, Forwards&&... forwardArgs)
        {
            return std::shared_ptr<ReadIntermediate<BodyT>>(
                new ReadIntermediate<BodyT>{*this, std::move(req), std::forward<Forwards>(forwardArgs)...});
        }
 
        template <typename BodyT = boost::beast::http::empty_body, typename RequestBodyT, typename... Forwards>
        [[nodiscard]] Response<BodyT> prepareResponse(Request<RequestBodyT> const& req, Forwards&&... forwardArgs)
        {
            auto res = Response<BodyT>{std::forward<Forwards>(forwardArgs)...};
            res.setHeader(boost::beast::http::field::server, "Roar+" BOOST_BEAST_VERSION_STRING);
            if (routeOptions().cors)
                res.enableCors(req, routeOptions().cors);
            return res;
        }
        template <typename BodyT = boost::beast::http::empty_body>
        [[nodiscard]] Response<BodyT> prepareResponse()
        {
            auto res = Response<BodyT>{};
            res.setHeader(boost::beast::http::field::server, "Roar+" BOOST_BEAST_VERSION_STRING);
            if (routeOptions().cors)
                res.enableCorsEverything();
            return res;
        }
 
        template <typename FunctionT>
        void withStreamDo(FunctionT&& func)
        {
            std::visit(std::forward<FunctionT>(func), stream());
        }
 
        [[nodiscard]] Detail::PromiseTypeBind<
            Detail::PromiseTypeBindThen<std::shared_ptr<WebsocketSession>>,
            Detail::PromiseTypeBindFail<Error const&>>
        upgrade(Request<boost::beast::http::empty_body> const& req);
 
        [[nodiscard]] RouteOptions const& routeOptions();
 
        bool isSecure() const;
 
        void sendStandardResponse(boost::beast::http::status status, std::string_view additionalInfo = "");
 
        void sendStrictTransportSecurityResponse();
 
        template <typename OriginalBodyT>
        Detail::PromiseTypeBind<Detail::PromiseTypeBindThen<>, Detail::PromiseTypeBindFail<Error const&>>
        awaitClientClose(Request<OriginalBodyT> req, std::chrono::milliseconds timeout = std::chrono::seconds{3})
        {
            return promise::newPromise([this, &req, &timeout](promise::Defer d) {
                read<VoidBody>(std::move(req))
                    ->useFixedTimeout(timeout)
                    .commit()
                    .then([d](auto&, auto const&) {
                        d.resolve();
                    })
                    .fail([d](Error const& e) {
                        d.reject(e);
                    });
            });
        }
 
      private:
        void setupRouteOptions(RouteOptions options);
 
      private:
        void readHeader();
        void performSslHandshake();
        bool onWriteComplete(bool expectsClose, boost::beast::error_code ec, std::size_t);
        std::variant<Detail::StreamType, boost::beast::ssl_stream<Detail::StreamType>>& stream();
        std::shared_ptr<boost::beast::http::request_parser<boost::beast::http::empty_body>>& parser();
        boost::beast::flat_buffer& buffer();
        StandardResponseProvider const& standardResponseProvider();
        void startup(bool immediate = true);
 
      private:
        struct Implementation;
        std::unique_ptr<Implementation> impl_;
    };
}