src/ceph/src/rgw/rgw_asio_client.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #ifndef RGW_ASIO_CLIENT_H
   4 #define RGW_ASIO_CLIENT_H
   5
   6 #include <boost/asio/ip/tcp.hpp>
   7 #include <beast/http/message.hpp>
   8 #include <beast/http/message_parser.hpp>
   9 #include <beast/core/flat_streambuf.hpp>
  10 #include "include/assert.h"
  11
  12 #include "rgw_client_io.h"
  13
  14 namespace rgw {
  15 namespace asio {
  16
  17 /// streaming message body interface
  18 struct streaming_body {
  19   using value_type = boost::asio::mutable_buffer;
  20
  21   class reader {
  22     value_type& buffer;
  23    public:
  24     using mutable_buffers_type = boost::asio::mutable_buffers_1;
  25
  26     static const bool is_direct{true}; // reads directly into user buffer
  27
  28     template<bool isRequest, class Fields>
  29     explicit reader(beast::http::message<isRequest, streaming_body, Fields>& m)
  30       : buffer(m.body)
  31     {}
  32
  33     void init() {}
  34     void init(uint64_t content_length) {}
  35     void finish() {}
  36
  37     mutable_buffers_type prepare(size_t n) {
  38       n = std::min(n, boost::asio::buffer_size(buffer));
  39       auto position = boost::asio::buffer_cast<char*>(buffer);
  40       return {position, n};
  41     }
  42
  43     void commit(size_t n) {
  44       buffer = buffer + n;
  45     }
  46   };
  47 };
  48
  49 using header_type = beast::http::fields;
  50 using parser_type = beast::http::message_parser<true, streaming_body, header_type>;
  51
  52 class ClientIO : public io::RestfulClient,
  53                  public io::BuffererSink {
  54  private:
  55   using tcp = boost::asio::ip::tcp;
  56   tcp::socket& socket;
  57   parser_type& parser;
  58   beast::flat_streambuf& buffer; //< parse buffer
  59
  60   bool conn_keepalive{false};
  61   bool conn_close{false};
  62   RGWEnv env;
  63
  64   rgw::io::StaticOutputBufferer<> txbuf;
  65
  66   size_t write_data(const char *buf, size_t len) override;
  67   size_t read_data(char *buf, size_t max);
  68
  69  public:
  70   ClientIO(tcp::socket& socket, parser_type& parser,
  71            beast::flat_streambuf& buffer);
  72   ~ClientIO() override;
  73
  74   bool get_conn_close() const { return conn_close; }
  75
  76   void init_env(CephContext *cct) override;
  77   size_t complete_request() override;
  78   void flush() override;
  79   size_t send_status(int status, const char *status_name) override;
  80   size_t send_100_continue() override;
  81   size_t send_header(const boost::string_ref& name,
  82                      const boost::string_ref& value) override;
  83   size_t send_content_length(uint64_t len) override;
  84   size_t complete_header() override;
  85
  86   size_t recv_body(char* buf, size_t max) override {
  87     return read_data(buf, max);
  88   }
  89
  90   size_t send_body(const char* buf, size_t len) override {
  91     return write_data(buf, len);
  92   }
  93
  94   RGWEnv& get_env() noexcept override {
  95     return env;
  96   }
  97 };
  98
  99 } // namespace asio
 100 } // namespace rgw
 101
 102 #endif // RGW_ASIO_CLIENT_H