Implement electrum protocol outpoint notifications.

Author: evoskuil

include/bitcoin/server/protocols/protocol.hpp

@@ -61,4 +61,7 @@ class BCS_API protocol
 } // namespace server
 } // namespace libbitcoin
 
+// For use with secondary (e.g. notification) strands.
+#define NOTIFY(method, ...) notify<CLASS>(&CLASS::method, __VA_ARGS__)
+
 #endif

include/bitcoin/server/protocols/protocol_electrum.hpp

@@ -247,34 +247,48 @@ class BCS_API protocol_electrum
     /// Address.
     /// -----------------------------------------------------------------------
 
-    // subscription.
+    // subscribe.
     void scripthash_subscribe(const hash_digest& hash,
         notify_t type) NOEXCEPT;
     void do_scripthash_subscribe(const hash_digest& hash,
         notify_t type) NOEXCEPT;
     void complete_scripthash_subscribe(const code& ec,
         hash_digest& status, const hash_digest& hash) NOEXCEPT;
 
-    // unsubscription.
+    // unsubscribe.
     void scripthash_unsubscribe(const hash_digest& hash) NOEXCEPT;
     void do_scripthash_unsubscribe(const hash_digest& hash) NOEXCEPT;
     void complete_scripthash_unsubscribe(bool found) NOEXCEPT;
 
-    // notification (do_scripthash()).
+    // notify (do_scripthash()).
     void scripthash_notify(const hash_digest& status, const hash_digest& hash,
         notify_t type) NOEXCEPT;
 
     /// Outpoint.
     /// -----------------------------------------------------------------------
 
-    // subscription (do_outpoint()).
+    // subscribe.
+    void do_outpoint_subscribe(const system::chain::point& prevout,
+        const std::string& hint) NOEXCEPT;
+    void complete_outpoint_subscribe(const code& ec,
+        const system::chain::point& prevout,
+        const std::string& hint) NOEXCEPT;
+
+    // unsubscribe.
+    void do_outpoint_unsubscribe(const system::chain::point& prevout) NOEXCEPT;
+    void complete_outpoint_unsubscribe(bool found) NOEXCEPT;
+
+    // notify (do_outpoint()).
+    void outpoint_notify(const std::unique_ptr<interface::object_t>& status,
+        const system::chain::point& prevout) NOEXCEPT;
+
+    // utility.
     bool get_outpoint_status(interface::object_t& status,
         const system::chain::point& prevout) const NOEXCEPT;
     bool send_outpoint_status(const system::chain::point& prevout,
-        const std::string& spk_hint) NOEXCEPT;
+        const std::string& hint) NOEXCEPT;
 
-    // unsubscription.
-    // notification.
+    // notify.
 
     /// Utilities.
     /// -----------------------------------------------------------------------
@@ -328,10 +342,10 @@ class BCS_API protocol_electrum
         const hash_digest& hash) NOEXCEPT;
 
     // Transformations.
-    static std::string to_method_name(notify_t type) NOEXCEPT;
     static array_t transform(const unspents& unspents) NOEXCEPT;
     static array_t transform(const histories& histories) NOEXCEPT;
     static hash_digest to_status(const histories& histories) NOEXCEPT;
+    static std::string to_method_name(notify_t type) NOEXCEPT;
 
     // Compute server.features.hosts value from config.
     object_t self_hosts() const NOEXCEPT;
@@ -365,8 +379,8 @@ class BCS_API protocol_electrum
     network::asio::strand notification_strand_;
 
     // These are protected by notification strand.
-    std::map<hash_digest, subscription> address_subscriptions_{};
     std::set<system::chain::point> outpoint_subscriptions_{};
+    std::map<hash_digest, subscription> address_subscriptions_{};
 };
 
 } // namespace server

src/protocols/electrum/protocol_electrum.cpp

@@ -28,11 +28,11 @@ namespace libbitcoin {
 namespace server {
 
 #define CLASS protocol_electrum
-#define NOTIFY(method, ...) notify<CLASS>(&CLASS::method, __VA_ARGS__)
 
 using namespace system;
 using namespace network::rpc;
 using namespace std::placeholders;
+constexpr auto relaxed = std::memory_order_relaxed;
 
 BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
 
@@ -137,7 +137,6 @@ bool protocol_electrum::handle_event(const code&, node::chase event_,
         return false;
 
     // TODO: collapse three atomics this into a single enumeration.
-    constexpr auto relaxed = std::memory_order_relaxed;
     switch (event_)
     {
         ////case node::chase::transaction:

src/protocols/electrum/protocol_electrum_outputs.cpp

@@ -27,6 +27,7 @@ namespace server {
 #define CLASS protocol_electrum
 
 using namespace system;
+using namespace system::chain;
 using namespace network::rpc;
 using namespace std::placeholders;
 constexpr auto relaxed = std::memory_order_relaxed;
@@ -104,10 +105,12 @@ void protocol_electrum::handle_blockchain_outpoint_get_status(const code& ec,
         return;
     }
 
-    chain::point prevout{ hash, index };
-    send_outpoint_status(prevout, spk_hint);
+    send_outpoint_status({ hash, index }, spk_hint);
 }
 
+// subscribe
+// ----------------------------------------------------------------------------
+
 void protocol_electrum::handle_blockchain_outpoint_subscribe(const code& ec,
     rpc_interface::blockchain_outpoint_subscribe, const std::string& tx_hash,
     double txout_idx, const std::string& spk_hint) NOEXCEPT
@@ -130,16 +133,54 @@ void protocol_electrum::handle_blockchain_outpoint_subscribe(const code& ec,
         return;
     }
 
-    // TODO: subscribe.
-    ///////////////////////////////////////////////////////////////////////////
-    subscribed_outpoint_.store(true, relaxed);
-    ///////////////////////////////////////////////////////////////////////////
+    // Outpoint status is not long-running so the notification strand is only
+    // used to guard the notifications set. No need for the monitor, as
+    // do_outpoint_subscribe() is trivial and pointless to cancel.
+    ////monitor(true);
+    NOTIFY(do_outpoint_subscribe, point{ hash, index }, spk_hint);
+}
+
+void protocol_electrum::do_outpoint_subscribe(const point& prevout,
+    const std::string& hint) NOEXCEPT
+{
+    // Cancellability is preserved because not on channel strand.
+    BC_ASSERT(notification_strand_.running_in_this_thread());
+
+    code ec{};
+    if (outpoint_subscriptions_.size() < options_.maximum_subscriptions)
+    {
+        // Subscription response is idempotent.
+        outpoint_subscriptions_.insert(prevout);
+        subscribed_outpoint_.store(true, relaxed);
+    }
+    else
+    {
+        ec = error::subscription_limit;
+    }
+
+    POST(complete_outpoint_subscribe, ec, prevout, hint);
+}
 
-    chain::point prevout{ hash, index };
-    if (!send_outpoint_status(prevout, spk_hint))
+void protocol_electrum::complete_outpoint_subscribe(const code& ec,
+    const point& prevout, const std::string& hint) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    ////monitor(false);
+    if (stopped())
         return;
+
+    if (ec)
+    {
+        send_code(ec);
+        return;
+    }
+
+    send_outpoint_status(prevout, hint);
 }
 
+// unsubscribe
+// ----------------------------------------------------------------------------
+
 void protocol_electrum::handle_blockchain_outpoint_unsubscribe(const code& ec,
     rpc_interface::blockchain_outpoint_unsubscribe, const std::string& tx_hash,
     double txout_idx) NOEXCEPT
@@ -162,51 +203,69 @@ void protocol_electrum::handle_blockchain_outpoint_unsubscribe(const code& ec,
         return;
     }
 
-    chain::point prevout{ hash, index };
+    NOTIFY(do_outpoint_unsubscribe, point{ hash, index });
+}
+
+void protocol_electrum::do_outpoint_unsubscribe(const point& prevout) NOEXCEPT
+{
+    BC_ASSERT(notification_strand_.running_in_this_thread());
 
-    // TODO: unsubscribe.
-    ///////////////////////////////////////////////////////////////////////////
-    const auto prior = subscribed_outpoint_.load(relaxed);
-    ///////////////////////////////////////////////////////////////////////////
+    const auto found = to_bool(outpoint_subscriptions_.erase(prevout));
+    if (is_zero(outpoint_subscriptions_.size()))
+        subscribed_outpoint_.store(false, relaxed);
 
-    send_result(prior, 16, BIND(complete, _1));
+    POST(complete_outpoint_unsubscribe, found);
 }
 
-// notification.
-// ============================================================================
+void protocol_electrum::complete_outpoint_unsubscribe(bool found) NOEXCEPT
+{
+    send_result(found, 16, BIND(complete, _1));
+}
+
+// notify
+// ----------------------------------------------------------------------------
 
 // Notifier for blockchain_outpoint_subscribe events.
-void protocol_electrum::do_outpoint(node::header_t link) NOEXCEPT
+void protocol_electrum::do_outpoint(node::header_t) NOEXCEPT
 {
+    // Cancellability is preserved because not on channel strand.
     BC_ASSERT(notification_strand_.running_in_this_thread());
 
-    // TODO: get prevout from event.
-    ///////////////////////////////////////////////////////////////////////////
-    chain::point prevout{};
-    ///////////////////////////////////////////////////////////////////////////
-
-    object_t status{};
-    if (!get_outpoint_status(status, prevout))
+    for (const auto& prevout: outpoint_subscriptions_)
     {
-        LOGF("Electrum::do_outpoint, outpoint not found (" << link << ").");
-        return;
+        if (stopping_)
+            return;
+
+        auto status = std::make_unique<object_t>();
+        if (!get_outpoint_status(*status, prevout))
+        {
+            LOGV("Electrum::do_outpoint, outpoint not found.");
+        }
+        else
+        {
+            // Asio-buffered message (small, not under caller control).
+            POST(outpoint_notify, std::move(status), prevout);
+        }
     }
+}
+
+void protocol_electrum::outpoint_notify(const std::unique_ptr<object_t>& status,
+    const point& prevout) NOEXCEPT
+{
+    BC_ASSERT(stranded());
 
-    // TODO: post_notification(bounce to network strand).
     send_notification("blockchain.outpoint.subscribe", array_t
     {
         array_t{ encode_hash(prevout.hash()), prevout.index() },
-        std::move(status)
+        std::move(*status)
     }, 128, BIND(handle_send, _1));
 }
 
-// ============================================================================
-
-// utility.
+// utility
 // ----------------------------------------------------------------------------
 
 bool protocol_electrum::get_outpoint_status(object_t& status,
-    const chain::point& prevout) const NOEXCEPT
+    const point& prevout) const NOEXCEPT
 {
     // May be on either network or notification strand (thread safe).
 
@@ -225,17 +284,17 @@ bool protocol_electrum::get_outpoint_status(object_t& status,
     return true;
 }
 
-bool protocol_electrum::send_outpoint_status(const chain::point& prevout,
-    const std::string& spk_hint) NOEXCEPT
+bool protocol_electrum::send_outpoint_status(const point& prevout,
+    const std::string& hint) NOEXCEPT
 {
     BC_ASSERT(stranded());
 
     // This is parsed for correctness but is not used.
     // Script is advisory, and should match output script.
-    if (!spk_hint.empty())
+    if (!hint.empty())
     {
         data_chunk bytes{};
-        if (!decode_base16(bytes, spk_hint))
+        if (!decode_base16(bytes, hint))
         {
             send_code(error::invalid_argument);
             return false;