Code review.

include/bitcoin/network/config/utilities.hpp

@@ -36,6 +36,10 @@ BCT_API asio::address from_address(const messages::peer::ip_address& address) NO
 BCT_API std::string to_normal_host(const std::string& host,
     uint16_t default_port) NOEXCEPT;
 
+/// Get ascii lowered host:port form of a url, with default substitution.
+BCT_API std::string to_normal_origin(const std::string& origin,
+    uint16_t default_port) NOEXCEPT;
+
 } // namespace config
 } // namespace network
 } // namespace libbitcoin

include/bitcoin/network/error.hpp

@@ -348,6 +348,7 @@ enum error_t : uint8_t
     missing_parameter,
 
     // json-rpc error
+    jsonrpc_invalid_version,
     jsonrpc_requires_method,
     jsonrpc_v1_requires_params,
     jsonrpc_v1_requires_array_params,

include/bitcoin/network/net.hpp

@@ -116,6 +116,9 @@ class BCT_API net
     /// Resume all P2P connection.
     virtual bool resume() NOEXCEPT;
 
+    /// Stop all service connections (not resumable).
+    virtual void stop_services(const code& ec) NOEXCEPT;
+
     /// Properties.
     /// -----------------------------------------------------------------------
 
@@ -207,6 +210,11 @@ class BCT_API net
         return stop_subscriber_.size();
     }
 
+    virtual size_t service_subscriber_count() const NOEXCEPT
+    {
+        return service_subscriber_.size();
+    }
+
     virtual size_t connect_subscriber_count() const NOEXCEPT
     {
         return connect_subscriber_.size();
@@ -241,6 +249,22 @@ class BCT_API net
         return session;
     }
 
+    /// Attach service session to network, caller must start (requires strand).
+    template <class Session, class Network, typename... Args>
+    typename Session::ptr attach_service(Network& net, Args&&... args) NOEXCEPT
+    {
+        BC_ASSERT_MSG(stranded(), "subscribe_service");
+        const auto session = attach<Session>(net, std::forward<Args>(args)...);
+
+        subscribe_service([=](const code&) NOEXCEPT
+        {
+            session->stop();
+            return false;
+        }, create_key());
+
+        return session;
+    }
+
     /// Base session (no derived session access).
     /// -----------------------------------------------------------------------
     friend class session;
@@ -324,6 +348,9 @@ class BCT_API net
     void do_subscribe_close(const stop_handler& handler,
         const stop_completer& complete) NOEXCEPT;
 
+    code subscribe_service(stop_handler&& handler, object_key key) NOEXCEPT;
+    void do_stop_services(const code& ec) NOEXCEPT;
+
     // P2P
     // ------------------------------------------------------------------------
 
@@ -373,6 +400,7 @@ class BCT_API net
     channel_ids actives_{};
     broadcaster broadcaster_{};
     stop_subscriber stop_subscriber_{};
+    stop_subscriber service_subscriber_{};
     channel_subscriber connect_subscriber_{};
     std::unordered_set<uint64_t> nonces_{};
 };

include/bitcoin/network/net/acceptor.hpp

@@ -100,6 +100,10 @@ class BCT_API acceptor
     // These are protected by strand.
     asio::acceptor acceptor_;
     bool stopped_{ true };
+
+private:
+    void do_handle_accept(const code& ec, const socket::ptr& socket,
+        const socket_handler& handler) NOEXCEPT;
 };
 
 } // namespace network

include/bitcoin/network/protocols/protocol_version_70002.hpp

@@ -42,7 +42,7 @@ class BCT_API protocol_version_70002
 
     /// Construct a version protocol instance using parameterized relay.
     protocol_version_70002(const session::ptr& session,
-        const channel::ptr& channel, bool relay) NOEXCEPT;
+        const channel::ptr& channel, bool relay, bool reject) NOEXCEPT;
 
     /// Perform the handshake (requires strand), handler invoked on completion.
     void shake(result_handler&& handle_event) NOEXCEPT override;
@@ -52,6 +52,10 @@ class BCT_API protocol_version_70002
 
     virtual bool handle_receive_reject(const code& ec,
         const messages::peer::reject::cptr& message) NOEXCEPT;
+
+private:
+    // This is thread safe.
+    const bool reject_;
 };
 
 } // namespace network

include/bitcoin/network/protocols/protocol_version_70014.hpp

@@ -52,10 +52,6 @@ class BCT_API protocol_version_70014
         const messages::peer::version_acknowledge::cptr& message) NOEXCEPT override;
     virtual bool handle_receive_send_compact(const code& ec,
         const messages::peer::send_compact::cptr& message) NOEXCEPT;
-
-private:
-    // This is thread safe.
-    const bool reject_;
 };
 
 } // namespace network

src/channels/channel.cpp

@@ -88,6 +88,7 @@ void channel::stopping(const code& ec) NOEXCEPT
 void channel::pause() NOEXCEPT
 {
     BC_ASSERT(stranded());
+    BC_ASSERT(!held_);
     stop_expiration();
     stop_inactivity();
 

src/config/credential.cpp

@@ -92,7 +92,7 @@ std::string credential::to_string() const NOEXCEPT
 std::istream& operator>>(std::istream& input, credential& argument) THROWS
 {
     std::string value{};
-    input >> value;
+    std::getline(input, value);
 
     // Methods are optional, so a credential has two or three tokens.
     const auto tokens = split(value, ":", true, false);

src/config/utilities.cpp

@@ -86,6 +86,20 @@ std::string to_normal_host(const std::string& host,
     }
 }
 
+std::string to_normal_origin(const std::string& origin,
+    uint16_t default_port) NOEXCEPT
+{
+    try
+    {
+        const url url{ origin };
+        return endpoint{ url.host(), url.port() }.to_lower(default_port);
+    }
+    catch (const std::exception&)
+    {
+        return {};
+    }
+}
+
 } // namespace config
 } // namespace network
 } // namespace libbitcoin

src/error.cpp

@@ -337,6 +337,7 @@ DEFINE_ERROR_T_MESSAGE_MAP(error)
     { missing_parameter, "missing parameter" },
 
     // json-rpc error
+    { jsonrpc_invalid_version, "jsonrpc invalid version" },
     { jsonrpc_requires_method, "jsonrpc requires method" },
     { jsonrpc_v1_requires_params, "jsonrpc v1 requires params" },
     { jsonrpc_v1_requires_array_params, "jsonrpc v1 requires array params" },

src/messages/http_body.cpp

@@ -40,6 +40,23 @@ void body::reader::init(const length_type& length, boost_code& ec) NOEXCEPT
         {
             ec = to_http_code(http_error_t::end_of_stream);
         },
+        [&](file_reader& read) NOEXCEPT
+        {
+            if (!value_.get<file_value>().is_open())
+            {
+                ec = to_errc_code(boost_errc_t::bad_file_descriptor);
+                return;
+            }
+
+            try
+            {
+                read.init(length, ec);
+            }
+            catch (...)
+            {
+                ec = to_http_code(http_error_t::end_of_stream);
+            }
+        },
         [&](auto& read) NOEXCEPT
         {
             try
@@ -132,16 +149,9 @@ body::writer::out_buffer body::writer::get(boost_code& ec) NOEXCEPT
     {
         [&] (std::monostate&) NOEXCEPT
         {
-            ec = {};
+            ec = to_http_code(http_error_t::end_of_stream);
             return out_buffer{};
         },
-        [&](empty_writer&) NOEXCEPT
-        {
-            ec = {};
-
-            // Socket body writer requires non-empty buffer to write empty.
-            return out_buffer{ std::make_pair(asio::const_buffer{}, false) };
-        },
         [&](auto& write) NOEXCEPT
         {
             try

src/messages/peer/detail/reject.cpp

@@ -90,26 +90,26 @@ reject reject::deserialize(uint32_t, reader& source) NOEXCEPT
     const auto code = byte_to_reason(source.read_byte());
     auto reason = source.read_string(max_reject_message);
 
-    const auto get_hash = [&message](reader& source) NOEXCEPT
+    auto hash = null_hash;
+    if (source && is_chain(message))
     {
-        if (!source || !is_chain(message))
-            return null_hash;
+        const auto position = source.get_read_position();
+        hash = source.read_hash();
 
         // Some nodes don't follow documented convention of hash for tx/block.
-        // Reset fault that occurs on hash read to accomodate those slackers.
-        const auto hash = source.read_hash();
         if (!source)
-            source.set_limit();
-
-        return hash;
-    };
+        {
+            source.set_position(position);
+            hash = null_hash;
+        }
+    }
 
     return
     {
         std::move(message),
         code,
         std::move(reason),
-        get_hash(source)
+        hash
     };
 }
 

src/messages/rpc/body.cpp

@@ -85,18 +85,24 @@ static void to_request(rpc::request& value, boost_code& ec) NOEXCEPT
     {
         // Primary exception type for parsing operations.
         ec = e.code();
+        return;
     }
     catch (...)
     {
         ec = code{ error::jsonrpc_reader_exception };
+        return;
     }
 
     // Set version default.
     if (value.message.jsonrpc == version::undefined)
         value.message.jsonrpc = version::v1;
 
     // Post-parse semantic validation.
-    if (value.message.method.empty())
+    if (value.message.jsonrpc == version::invalid)
+    {
+        ec = code{ error::jsonrpc_invalid_version };
+    }
+    else if (value.message.method.empty())
     {
         ec = code{ error::jsonrpc_requires_method };
     }

src/messages/rpc/model.cpp

@@ -130,6 +130,10 @@ DEFINE_JSON_FROM_TAG(value_t)
             auto& object = value.as_object();
             for (const auto& pair: visit)
                 object.emplace(pair.first, value_from(pair.second));
+        },
+        [&](const any_t&) THROWS
+        {
+            throw ostream_exception{ "any_t" };
         }
     }, instance.value());
 }
@@ -381,7 +385,8 @@ DEFINE_JSON_TO_TAG(response_t)
     }
 
     // error
-    if (const auto it = object.find("error"); it != object.end())
+    if (const auto it = object.find("error");
+        it != object.end() && !it->value().is_null())
     {
         result_t result{};
         const auto& error = it->value().as_object();

src/net.cpp

@@ -301,6 +301,7 @@ void net::do_close() NOEXCEPT
 
     // Notify and delete all stop subscribers (all sessions).
     stop_subscriber_.stop(error::service_stopped);
+    service_subscriber_.stop(error::service_stopped);
 
     // Notify and delete subscribers to channel notifications.
     connect_subscriber_.stop_default(error::service_stopped);
@@ -344,6 +345,19 @@ bool net::resume() NOEXCEPT
     return true;
 }
 
+void net::stop_services(const code& ec) NOEXCEPT
+{
+    boost::asio::post(strand_,
+        std::bind(&net::do_stop_services, this, ec));
+}
+
+void net::do_stop_services(const code& ec) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    suspend_services();
+    service_subscriber_.stop(ec);
+}
+
 bool net::suspended() const NOEXCEPT
 {
     return connect_suspended_.load() || accept_suspended_.load();
@@ -466,6 +480,12 @@ code net::subscribe_close(stop_handler&& handler, object_key key) NOEXCEPT
     return stop_subscriber_.subscribe(std::move(handler), key);
 }
 
+code net::subscribe_service(stop_handler&& handler, object_key key) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+    return service_subscriber_.subscribe(std::move(handler), key);
+}
+
 // public
 void net::subscribe_close(stop_handler&& handler,
     stop_completer&& complete) NOEXCEPT

src/net/acceptor.cpp

@@ -164,7 +164,7 @@ void acceptor::accept(socket_handler&& handler) NOEXCEPT
             shared_from_this(), _1, socket, std::move(handler)));
 }
 
-// private
+// protected
 void acceptor::handle_accept(const code& ec, const socket::ptr& socket,
     const socket_handler& handler) NOEXCEPT
 {
@@ -173,11 +173,20 @@ void acceptor::handle_accept(const code& ec, const socket::ptr& socket,
     if (!stranded())
     {
         boost::asio::post(strand_,
-            std::bind(&acceptor::handle_accept,
+            std::bind(&acceptor::do_handle_accept,
                 shared_from_this(), ec, socket, handler));
         return;
     }
 
+    do_handle_accept(ec, socket, handler);
+}
+
+// private
+void acceptor::do_handle_accept(const code& ec, const socket::ptr& socket,
+    const socket_handler& handler) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+
     if (ec)
     {
         socket->stop();