Generalize channel_rpc/protocol_rpc for notification write.

Author: evoskuil

include/bitcoin/network/channels/channel_rpc.hpp

@@ -64,16 +64,31 @@ class channel_rpc
         result_handler&& handler) NOEXCEPT;
     inline void send_result(rpc::value_t&& result, size_t size_hint,
         result_handler&& handler) NOEXCEPT;
+    inline void send_notification(const rpc::string_t& method,
+        rpc::params_t&& notification, size_t size_hint,
+        result_handler&& handler) NOEXCEPT;
 
     /// Resume reading from the socket (requires strand).
     inline void resume() NOEXCEPT override;
 
 protected:
     /// Serialize and write response to client (requires strand).
     /// Completion handler is always invoked on the channel strand.
-    inline void send(rpc::response_t&& message, size_t size_hint,
+    template <typename Message>
+    inline void send(Message&& message, size_t size_hint,
         result_handler&& handler) NOEXCEPT;
 
+    /// Size and assign response_buffer_ (value type is json-rpc::json).
+    template <typename Message>
+    inline rpc::message_ptr<Message> assign_message(Message&& message,
+        size_t size_hint) NOEXCEPT;
+
+    /// Handle send<response> completion, invokes receive().
+    template <typename Message>
+    inline void handle_send(const code& ec, size_t bytes,
+        const rpc::message_cptr<Message>& message,
+        const result_handler& handler) NOEXCEPT;
+
     /// Stranded handler invoked from stop().
     inline void stopping(const code& ec) NOEXCEPT override;
 
@@ -83,22 +98,13 @@ class channel_rpc
     /// Override to dispatch request to subscribers by requested method.
     virtual inline void dispatch(const rpc::request_cptr& request) NOEXCEPT;
 
-    /// Size and assign response_buffer_ (value type is json-rpc::json).
-    virtual inline rpc::response_ptr assign_message(rpc::response_t&& message,
-        size_t size_hint) NOEXCEPT;
-
     /// Must call after successful message handling if no stop.
     virtual inline void receive() NOEXCEPT;
 
     /// Handle incoming messages.
     virtual inline void handle_receive(const code& ec, size_t bytes,
         const rpc::request_cptr& request) NOEXCEPT;
 
-    /// Handle send completion, invokes receive().
-    virtual inline void handle_send(const code& ec, size_t bytes,
-        const rpc::response_cptr& response,
-        const result_handler& handler) NOEXCEPT;
-
 private:
     // These are protected by strand.
     rpc::version version_;

include/bitcoin/network/impl/channels/channel_rpc.ipp

@@ -140,40 +140,15 @@ inline http::flat_buffer& CLASS::request_buffer() NOEXCEPT
 // Send.
 // ----------------------------------------------------------------------------
 
-TEMPLATE
-void CLASS::send_code(const code& ec, result_handler&& handler) NOEXCEPT
-{
-    send_error({ .code = ec.value(), .message = ec.message() },
-        std::move(handler));
-}
-
-TEMPLATE
-void CLASS::send_error(rpc::result_t&& error,
-    result_handler&& handler) NOEXCEPT
-{
-    BC_ASSERT(stranded());
-    const auto hint = two * error.message.size();
-    send({ .jsonrpc = version_, .id = identity_, .error = std::move(error) },
-        hint, std::move(handler));
-}
-
-TEMPLATE
-void CLASS::send_result(rpc::value_t&& result, size_t size_hint,
-    result_handler&& handler) NOEXCEPT
-{
-    BC_ASSERT(stranded());
-    send({ .jsonrpc = version_, .id = identity_, .result = std::move(result) },
-        size_hint, std::move(handler));
-}
-
 // protected
 TEMPLATE
-inline void CLASS::send(rpc::response_t&& model, size_t size_hint,
+template <typename Message>
+inline void CLASS::send(Message&& model, size_t size_hint,
     result_handler&& handler) NOEXCEPT
 {
     BC_ASSERT(stranded());
     const auto out = assign_message(std::move(model), size_hint);
-    count_handler complete = std::bind(&CLASS::handle_send,
+    count_handler complete = std::bind(&CLASS::handle_send<Message>,
         shared_from_base<CLASS>(), _1, _2, out, std::move(handler));
 
     if (!out)
@@ -187,33 +162,96 @@ inline void CLASS::send(rpc::response_t&& model, size_t size_hint,
 
 // protected
 TEMPLATE
+template <typename Message>
+inline rpc::message_ptr<Message> CLASS::assign_message(Message&& message,
+    size_t size_hint) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    response_buffer_->max_size(size_hint);
+    const auto ptr = system::to_shared<rpc::message_value<Message>>();
+    ptr->message = std::move(message);
+    ptr->buffer = response_buffer_;
+    return ptr;
+}
+
+// protected
+TEMPLATE
+template <typename Message>
 inline void CLASS::handle_send(const code& ec, size_t bytes,
-    const rpc::response_cptr& response, const result_handler& handler) NOEXCEPT
+    const rpc::message_cptr<Message>& message,
+    const result_handler& handler) NOEXCEPT
 {
     BC_ASSERT(stranded());
     if (ec) stop(ec);
 
     // Typically a noop, but handshake may pause channel here.
     handler(ec);
 
-    LOGA("Rpc response: (" << bytes << ") bytes [" << endpoint() << "] "
-        << response->message.error.value_or(rpc::result_t{}).message);
+    if constexpr (is_same_type<Message, rpc::response_t>)
+    {
+        LOGA("Rpc response: (" << bytes << ") bytes [" << endpoint() << "] "
+            << message->message.error.value_or(rpc::result_t{}).message);
 
-    // Continue read loop (does not unpause or restart channel).
-    receive();
+        // Continue the read loop (does not unpause or restart).
+        receive();
+    }
+    else
+    {
+        LOGA("Rpc notification: (" << bytes << ") bytes [" << endpoint() << "] "
+            << message->message.method);
+    }
 }
 
-// private
 TEMPLATE
-inline rpc::response_ptr CLASS::assign_message(rpc::response_t&& message,
-    size_t size_hint) NOEXCEPT
+inline void CLASS::send_code(const code& ec, result_handler&& handler) NOEXCEPT
+{
+    send_error(
+    {
+        .code = ec.value(),
+        .message = ec.message()
+    },
+    std::move(handler));
+}
+
+TEMPLATE
+inline void CLASS::send_error(rpc::result_t&& error,
+    result_handler&& handler) NOEXCEPT
 {
     BC_ASSERT(stranded());
-    response_buffer_->max_size(size_hint);
-    const auto ptr = system::to_shared<rpc::response>();
-    ptr->message = std::move(message);
-    ptr->buffer = response_buffer_;
-    return ptr;
+    const auto hint = two * error.message.size();
+    send(rpc::response_t
+    {
+        .jsonrpc = version_,
+        .id = identity_,
+        .error = std::move(error)
+    }, hint, std::move(handler));
+}
+
+TEMPLATE
+inline void CLASS::send_result(rpc::value_t&& result, size_t size_hint,
+    result_handler&& handler) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    send(rpc::response_t
+    {
+        .jsonrpc = version_,
+        .id = identity_,
+        .result = std::move(result)
+    }, size_hint, std::move(handler));
+}
+
+TEMPLATE
+inline void CLASS::send_notification(const rpc::string_t& method,
+    rpc::params_t&& notification, size_t size_hint,
+    result_handler&& handler) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    send(rpc::request_t
+    {
+        .jsonrpc = version_,
+        .method = method,
+        .params = std::move(notification)
+    }, size_hint, std::move(handler));
 }
 
 BC_POP_WARNING()

include/bitcoin/network/impl/protocols/protocol_rpc.ipp

@@ -69,6 +69,15 @@ inline void CLASS::send_result(rpc::value_t&& result, size_t size_hint,
     channel_->send_result(std::move(result), size_hint, std::move(handler));
 }
 
+TEMPLATE
+inline void CLASS::send_notification(const rpc::string_t& method,
+    rpc::params_t&& notification, size_t size_hint,
+    result_handler&& handler) NOEXCEPT
+{
+    channel_->send_notification(method, std::move(notification), size_hint,
+        std::move(handler));
+}
+
 } // namespace network
 } // namespace libbitcoin
 

include/bitcoin/network/protocols/protocol_rpc.hpp

@@ -61,6 +61,9 @@ class protocol_rpc
         result_handler&& handler) NOEXCEPT;
     virtual inline void send_result(rpc::value_t&& result, size_t size_hint,
         result_handler&& handler) NOEXCEPT;
+    virtual inline void send_notification(const rpc::string_t& method, 
+        rpc::params_t&& notification, size_t size_hint,
+        result_handler&& handler) NOEXCEPT;
 
     /// Default noop completion handler.
     virtual inline void complete(const code&) NOEXCEPT {};