Implement json-rpc over tcp socket/proxy methods.

Author: evoskuil

include/bitcoin/network/error.hpp

@@ -238,7 +238,7 @@ BCT_API bool asio_is_canceled(const boost_code& ec) NOEXCEPT;
 BCT_API code asio_to_error_code(const boost_code& ec) NOEXCEPT;
 
 /// 1:1 mapping of boost::beast:http::error to network (or error::unknown).
-BCT_API code beast_to_error_code(const boost_code& ec) NOEXCEPT;
+BCT_API code http_to_error_code(const boost_code& ec) NOEXCEPT;
 
 } // namespace error
 } // namespace network

include/bitcoin/network/impl/messages/json_body.ipp

@@ -114,6 +114,12 @@ void CLASS::reader::finish(boost_code& ec) NOEXCEPT
     parser_.reset();
 }
 
+TEMPLATE
+bool CLASS::reader::done() const NOEXCEPT
+{
+    return parser_.done();
+}
+
 // json::body<>::writer
 // ----------------------------------------------------------------------------
 

include/bitcoin/network/messages/json_body.hpp

@@ -73,6 +73,7 @@ struct body
         virtual void init(const http::length_type& length, boost_code& ec) NOEXCEPT;
         virtual size_t put(const buffer_type& buffer, boost_code& ec) NOEXCEPT;
         virtual void finish(boost_code& ec) NOEXCEPT;
+        virtual bool done() const NOEXCEPT;
 
     protected:
         value_type& value_;
@@ -86,6 +87,7 @@ struct body
     class writer
     {
     public:
+        static constexpr size_t default_buffer = 4096;
         using const_buffers_type = asio::const_buffer;
         using out_buffer = http::get_buffer<const_buffers_type>;
 
@@ -108,9 +110,6 @@ struct body
     protected:
         value_type& value_;
         boost::json::serializer serializer_;
-
-    private:
-        static constexpr size_t default_buffer = 4096;
     };
 };
 

include/bitcoin/network/net/proxy.hpp

@@ -42,6 +42,8 @@ class BCT_API proxy
 public:
     typedef std::shared_ptr<proxy> ptr;
     typedef subscriber<> stop_subscriber;
+    typedef rpc::request_body::value_type rpc_in_value;
+    typedef rpc::response_body::value_type rpc_out_value;
 
     DELETE_COPY_MOVE(proxy);
 
@@ -126,19 +128,17 @@ class BCT_API proxy
         count_handler&& handler) NOEXCEPT;
 
     /// Send a complete TCP message to the remote endpoint.
-    virtual void write(const asio::const_buffer& payload,
+    virtual void write(const asio::const_buffer& buffer,
         count_handler&& handler) NOEXCEPT;
 
     /// TCP-RPC (e.g. electrum, stratum_v1).
     /// -----------------------------------------------------------------------
 
     /// Read full rpc request from the socket, handler posted to socket strand.
-    virtual void read(rpc::response_t& out,
-        count_handler&& handler) NOEXCEPT;
+    virtual void read(rpc_in_value& out, count_handler&& handler) NOEXCEPT;
 
     /// Write full rpc response to the socket, handler posted to socket strand.
-    virtual void write(const rpc::response_t& in,
-        count_handler&& handler) NOEXCEPT;
+    virtual void write(rpc_out_value&& in, count_handler&& handler) NOEXCEPT;
 
     /// HTTP (generic).
     /// -----------------------------------------------------------------------

include/bitcoin/network/net/socket.hpp

@@ -41,6 +41,8 @@ class BCT_API socket
 {
 public:
     typedef std::shared_ptr<socket> ptr;
+    typedef rpc::request_body::value_type rpc_in_value;
+    typedef rpc::response_body::value_type rpc_out_value;
 
     DELETE_COPY_MOVE(socket);
 
@@ -104,11 +106,11 @@ class BCT_API socket
     /// -----------------------------------------------------------------------
 
     /// Read full rpc request from the socket, handler posted to socket strand.
-    virtual void rpc_read(rpc::response_t& out,
+    virtual void rpc_read(rpc_in_value& request,
         count_handler&& handler) NOEXCEPT;
 
     /// Write full rpc response to the socket, handler posted to socket strand.
-    virtual void rpc_write(const rpc::response_t& model,
+    virtual void rpc_write(rpc_out_value&& response,
         count_handler&& handler) NOEXCEPT;
 
     /// HTTP (generic).
@@ -158,13 +160,46 @@ class BCT_API socket
     /// The socket was upgraded to a websocket (requires strand).
     virtual bool websocket() const NOEXCEPT;
 
+    /// Utility.
+    asio::socket& get_transport() NOEXCEPT;
+    void logx(const std::string& context, const boost_code& ec) const NOEXCEPT;
+
 private:
+    struct read_rpc
+    {
+        typedef std::shared_ptr<read_rpc> ptr;
+        using rpc_reader = rpc::request_body::reader;
+
+        read_rpc(rpc_in_value& request_) NOEXCEPT
+          : value{}, reader{ value }
+        {
+            request_ = value;
+        }
+
+        rpc_in_value value;
+        rpc_reader reader;
+    };
+
+    struct write_rpc
+    {
+        typedef std::shared_ptr<write_rpc> ptr;
+        using rpc_writer = rpc::response_body::writer;
+        using out_buffer = rpc_writer::out_buffer;
+
+        write_rpc(rpc_out_value&& response) NOEXCEPT
+          : value{ std::move(response) }, writer{ value }
+        {
+        }
+
+        rpc_out_value value;
+        rpc_writer writer;
+    };
+
     // stop
     // ------------------------------------------------------------------------
 
     void do_stop() NOEXCEPT;
     void do_async_stop() NOEXCEPT;
-    asio::socket& get_transport() NOEXCEPT;
 
     // wait
     // ------------------------------------------------------------------------
@@ -186,11 +221,9 @@ class BCT_API socket
         const count_handler& handler) NOEXCEPT;
 
     // tcp (rpc)
-    void do_rpc_read(
-        std::reference_wrapper<rpc::response_t> out,
+    void do_rpc_read(boost_code ec, size_t total, const read_rpc::ptr& in,
         const count_handler& handler) NOEXCEPT;
-    void do_rpc_write(
-        const std::reference_wrapper<const rpc::response_t>& in,
+    void do_rpc_write(boost_code ec, size_t total, const write_rpc::ptr& out,
         const count_handler& handler) NOEXCEPT;
 
     // http (generic)
@@ -228,8 +261,10 @@ class BCT_API socket
         const count_handler& handler) NOEXCEPT;
 
     // tcp (rpc)
-    void handle_rpc_tcp(const boost_code& ec, size_t size,
-        const count_handler& handler) NOEXCEPT;
+    void handle_rpc_read(boost_code ec, size_t size, size_t total,
+        const read_rpc::ptr& in, const count_handler& handler) NOEXCEPT;
+    void handle_rpc_write(boost_code ec, size_t size, size_t total,
+        const write_rpc::ptr& out, const count_handler& handler) NOEXCEPT;
 
     // http (generic)
     void handle_http_read(const boost_code& ec, size_t size,

src/error.cpp

@@ -316,7 +316,7 @@ code asio_to_error_code(const boost_code& ec) NOEXCEPT
     return error::unknown;
 }
 
-code beast_to_error_code(const boost_code& ec) NOEXCEPT
+code http_to_error_code(const boost_code& ec) NOEXCEPT
 {
     namespace beast = boost::beast::http;
 

src/net/proxy.cpp

@@ -194,28 +194,28 @@ void proxy::read(const asio::mutable_buffer& buffer,
     socket_->read(buffer, std::move(handler));
 }
 
-void proxy::write(const asio::const_buffer& payload,
+void proxy::write(const asio::const_buffer& buffer,
     count_handler&& handler) NOEXCEPT
 {
     boost::asio::dispatch(strand(),
         std::bind(&proxy::do_write,
-            shared_from_this(), payload, std::move(handler)));
+            shared_from_this(), buffer, std::move(handler)));
 }
 
 // TCP-RPC.
 // ----------------------------------------------------------------------------
 
-void proxy::read(rpc::response_t& out, count_handler&& handler) NOEXCEPT
+void proxy::read(rpc_in_value& out, count_handler&& handler) NOEXCEPT
 {
     boost::asio::dispatch(strand(),
         std::bind(&proxy::waiting, shared_from_this()));
 
     socket_->rpc_read(out, std::move(handler));
 }
 
-void proxy::write(const rpc::response_t& in, count_handler&& handler) NOEXCEPT
+void proxy::write(rpc_out_value&& in, count_handler&& handler) NOEXCEPT
 {
-    socket_->rpc_write(in, std::move(handler));
+    socket_->rpc_write(std::move(in), std::move(handler));
 
     // TODO: compose?
     ////boost::asio::dispatch(strand(),