ws_session.hpp

Go to the documentation of this file.

#ifndef KAGOME_CORE_API_TRANSPORT_IMPL_WS_SESSION_HPP
#define KAGOME_CORE_API_TRANSPORT_IMPL_WS_SESSION_HPP

#include <chrono>
#include <cstdlib>
#include <memory>
#include <queue>

#include <boost/asio/strand.hpp>
#include <boost/beast/core/multi_buffer.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <boost/beast/websocket.hpp>

#include "api/transport/session.hpp"
#include "log/logger.hpp"

namespace kagome::api {

  class WsSession : public Session,
                    public std::enable_shared_from_this<WsSession> {
    using WsError = boost::beast::websocket::error;
    using OnWsSessionCloseHandler = std::function<void()>;

   public:
    struct Configuration {
      static constexpr size_t kDefaultRequestSize = 10000u;
      static constexpr Duration kDefaultTimeout = std::chrono::seconds(30);

      size_t max_request_size{kDefaultRequestSize};
      Duration operation_timeout{kDefaultTimeout};
    };

    ~WsSession() override = default;

    WsSession(Context &context, Configuration config, SessionId id);

    Socket &socket() override {
      return socket_;
    }

    void start() override;

    Session::SessionId id() const override;

    SessionType type() const override {
      return SessionType::kWs;
    }

    void respond(std::string_view response) override;

    void reject();

    void connectOnWsSessionCloseHandler(OnWsSessionCloseHandler &&handler);

   private:
    void stop();

    void stop(boost::beast::websocket::close_code code);

    void handleRequest(std::string_view data);

    void asyncRead();

    void asyncWrite();

    void onRun();

    void onAccept(boost::system::error_code ec);

    void onRead(boost::system::error_code ec, std::size_t size);

    void onWrite(boost::system::error_code ec, std::size_t bytes_transferred);

    void reportError(boost::system::error_code ec, std::string_view message);

    boost::asio::strand<boost::asio::io_context::executor_type> strand_;

    boost::asio::ip::tcp::socket socket_;

    Configuration config_;  
    boost::beast::websocket::stream<boost::asio::ip::tcp::socket &> stream_;
    boost::beast::flat_buffer rbuffer_;  
    boost::beast::flat_buffer wbuffer_;  

    std::mutex cs_;
    std::queue<std::string> pending_responses_;

    std::atomic_bool writing_in_progress_ = false;
    std::atomic_bool stopped_ = false;

    SessionId const id_;
    OnWsSessionCloseHandler on_ws_close_;
    log::Logger logger_ = log::createLogger("WsSession", "rpc_transport");
  };

}  // namespace kagome::api

#endif  // KAGOME_CORE_API_TRANSPORT_BEAST_HTTP_SESSION_HPP