TASK00 - Авторизация и регистрация

src/session/HttpSession.cpp

@@ -0,0 +1,98 @@
+#include <boost/assert.hpp>
+
+#include "HttpSession.h"
+#include "WebsocketSession.h"
+#include "./../endpoints_handlers/HandleRequest.h"
+
+namespace uad
+{
+HttpSession::Queue::Queue(HttpSession& self) : self_(self)
+{
+  static_assert(limit > 0, "Queue limit must be positive");
+  items_.reserve(limit);
+}
+
+bool HttpSession::Queue::IsFull() const { return items_.size() >= limit; }
+
+bool HttpSession::Queue::OnWrite()
+{
+  BOOST_ASSERT(!items_.empty());
+  auto const was_full = IsFull();
+  items_.erase(items_.begin());
+  if (!items_.empty())
+    (*items_.front())();
+  return was_full;
+}
+
+HttpSession::HttpSession(boost::asio::ip::tcp::socket&& socket,
+                         std::shared_ptr<std::string const> const& doc_root) :
+    stream_(std::move(socket)), doc_root_(doc_root), queue_(*this)
+{
+}
+
+void HttpSession::Run()
+{
+  boost::asio::dispatch(
+    stream_.get_executor(),
+    boost::beast::bind_front_handler(&HttpSession::DoRead, this->shared_from_this()));
+}
+
+void HttpSession::DoRead()
+{
+  parser_.emplace();
+
+  parser_->body_limit(10000);
+
+  stream_.expires_after(std::chrono::seconds(30));
+
+  boost::beast::http::async_read(
+    stream_, buffer_, *parser_,
+    boost::beast::bind_front_handler(&HttpSession::OnRead, shared_from_this()));
+}
+
+void HttpSession::OnRead(boost::beast::error_code ec, std::size_t bytes_transferred)
+{
+  boost::ignore_unused(bytes_transferred);
+
+  if (ec == boost::beast::http::error::end_of_stream)
+    return DoClose();
+
+  if (ec)
+    return Fail(ec, "read");
+
+  if (boost::beast::websocket::is_upgrade(parser_->get()))
+  {
+    std::make_shared<WebsocketSession>(stream_.release_socket())->DoAccept(parser_->release());
+    return;
+  }
+
+  HandleRequest(*doc_root_, parser_->release(), queue_);
+
+  if (!queue_.IsFull())
+    DoRead();
+}
+
+void HttpSession::OnWrite(bool close, boost::beast::error_code ec, std::size_t bytes_transferred)
+{
+  boost::ignore_unused(bytes_transferred);
+
+  if (ec)
+    return Fail(ec, "write");
+
+  if (close)
+  {
+    return DoClose();
+  }
+
+  if (queue_.OnWrite())
+  {
+    DoRead();
+  }
+}
+
+void HttpSession::DoClose()
+{
+  boost::beast::error_code ec;
+  stream_.socket().shutdown(boost::asio::ip::tcp::socket::shutdown_send, ec);
+}
+} // namespace uad

src/session/HttpSession.h

@@ -0,0 +1,82 @@
+#pragma once
+
+#include <memory>
+#include <boost/beast.hpp>
+#include <boost/asio.hpp>
+
+namespace uad
+{
+class HttpSession : public std::enable_shared_from_this<HttpSession>
+{
+  class Queue
+  {
+    enum
+    {
+      limit = 8
+    };
+
+    struct work
+    {
+      virtual void operator()() = 0;
+      virtual ~work() = default;
+    };
+
+    HttpSession& self_;
+    std::vector<std::unique_ptr<work>> items_;
+
+  public:
+    explicit Queue(HttpSession& self);
+
+    bool IsFull() const;
+
+    bool OnWrite();
+
+    template <bool isRequest, class Body, class Fields>
+    void operator()(boost::beast::http::message<isRequest, Body, Fields>&& msg)
+    {
+      struct work_impl : work
+      {
+        HttpSession& self_;
+        boost::beast::http::message<isRequest, Body, Fields> msg_;
+
+        work_impl(HttpSession& self, boost::beast::http::message<isRequest, Body, Fields>&& msg) :
+            self_(self), msg_(std::move(msg))
+        {
+        }
+
+        void operator()()
+        {
+          boost::beast::http::async_write(self_.stream_, msg_,
+                            boost::beast::bind_front_handler(&HttpSession::OnWrite,
+                                                      self_.shared_from_this(), msg_.need_eof()));
+        }
+      };
+
+      items_.push_back(boost::make_unique<work_impl>(self_, std::move(msg)));
+
+      if (items_.size() == 1)
+        (*items_.front())();
+    }
+  };
+
+  boost::beast::tcp_stream stream_;
+  boost::beast::flat_buffer buffer_;
+  std::shared_ptr<std::string const> doc_root_;
+  Queue queue_;
+  boost::optional<boost::beast::http::request_parser<boost::beast::http::string_body>> parser_;
+
+public:
+  HttpSession(boost::asio::ip::tcp::socket&& socket, std::shared_ptr<std::string const> const& doc_root);
+
+  void Run();
+
+private:
+  void DoRead();
+
+  void OnRead(boost::beast::error_code ec, std::size_t bytes_transferred);
+
+  void OnWrite(bool close, boost::beast::error_code ec, std::size_t bytes_transferred);
+
+  void DoClose();
+};
+}

src/session/WebsocketSession.cpp

@@ -0,0 +1,51 @@
+#include "WebsocketSession.h"
+
+#include "./../helpers/helpers.h"
+
+namespace uad
+{
+WebsocketSession::WebsocketSession(boost::asio::ip::tcp::socket&& socket) : ws_(std::move(socket))
+{
+}
+
+void WebsocketSession::OnAccept(boost::beast::error_code ec)
+{
+  if (ec)
+    return Fail(ec, "accept");
+
+  DoRead();
+}
+
+void WebsocketSession::DoRead()
+{
+  ws_.async_read(buffer_,
+                 boost::beast::bind_front_handler(&WebsocketSession::OnRead, shared_from_this()));
+}
+
+void WebsocketSession::OnRead(boost::beast::error_code ec, std::size_t bytes_transferred)
+{
+  boost::ignore_unused(bytes_transferred);
+
+  if (ec == boost::beast::websocket::error::closed)
+    return;
+
+  if (ec)
+    Fail(ec, "read");
+
+  ws_.text(ws_.got_text());
+  ws_.async_write(buffer_.data(),
+                  boost::beast::bind_front_handler(&WebsocketSession::OnWrite, shared_from_this()));
+}
+
+void WebsocketSession::OnWrite(boost::beast::error_code ec, std::size_t bytes_transferred)
+{
+  boost::ignore_unused(bytes_transferred);
+
+  if (ec)
+    return Fail(ec, "write");
+
+  buffer_.consume(buffer_.size());
+
+  DoRead();
+}
+} // namespace uad

src/session/WebsocketSession.h

@@ -0,0 +1,41 @@
+#pragma once
+
+#include <memory>
+#include <boost/beast.hpp>
+#include <boost/asio.hpp>
+
+namespace uad
+{
+class WebsocketSession : public std::enable_shared_from_this<WebsocketSession>
+{
+  boost::beast::websocket::stream<boost::beast::tcp_stream> ws_;
+  boost::beast::flat_buffer buffer_;
+
+public:
+  explicit WebsocketSession(boost::asio::ip::tcp::socket&& socket);
+
+  template <class Body, class Allocator>
+  void DoAccept(boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>> req)
+  {
+    ws_.set_option(boost::beast::websocket::stream_base::timeout::suggested(boost::beast::role_type::server));
+
+    ws_.set_option(boost::beast::websocket::stream_base::decorator(
+      [](boost::beast::websocket::response_type& res)
+      {
+        res.set(boost::beast::http::field::server, std::string(BOOST_BEAST_VERSION_STRING) + " advanced-server");
+      }));
+
+    ws_.async_accept(req,
+                     boost::beast::bind_front_handler(&WebsocketSession::OnAccept, shared_from_this()));
+  }
+
+private:
+  void OnAccept(boost::beast::error_code ec);
+
+  void DoRead();
+
+  void OnRead(boost::beast::error_code ec, std::size_t bytes_transferred);
+
+  void OnWrite(boost::beast::error_code ec, std::size_t bytes_transferred);
+};
+}