Fix subscription, add OpenBSD to build, use config settings.

include/bitcoin/server/server_node.hpp

@@ -69,13 +69,12 @@ class BCS_API server_node
         const incoming_message& request, queue_send_callback queue_send);
 
 protected:
-    // Result of store operation in transaction pool.
-    virtual void new_unconfirm_valid_tx(const std::error_code& ec,
-        const index_list& unconfirmed, const transaction_type& tx);
+    bool handle_tx(const std::error_code& ec, const index_list& unconfirmed,
+        const transaction_type& tx);
 
-    virtual void broadcast_new_blocks(const std::error_code& ec,
-        uint32_t fork_point, const chain::blockchain::block_list& new_blocks,
-        const chain::blockchain::block_list& replaced_blocks);
+    bool handle_reorg(const std::error_code& ec, uint64_t fork_point,
+        const chain::blockchain::block_list& new_blocks,
+        const chain::blockchain::block_list&);
 
 private:
     typedef std::vector<block_notify_callback> block_notify_list;

include/bitcoin/server/subscribe_manager.hpp

@@ -22,6 +22,7 @@
 
 #include <cstdint>
 #include <unordered_map>
+#include <vector>
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <bitcoin/bitcoin.hpp>
 #include <bitcoin/server/define.hpp>
@@ -41,7 +42,7 @@ enum class subscribe_type
 class BCS_API subscribe_manager
 {
 public:
-    subscribe_manager(server_node& node,
+    subscribe_manager(server_node& server,
         uint32_t maximum_subscriptions=100000000,
         uint32_t subscription_expiration_minutes=10);
 
@@ -74,11 +75,16 @@ class BCS_API subscribe_manager
         queue_send_callback queue_send);
     void do_submit(size_t height, const hash_digest& block_hash,
         const transaction_type& tx);
-    void post_updates(const payment_address& address,
-        size_t height, const hash_digest& block_hash,
-        const transaction_type& tx);
-    void post_stealth_updates(const binary_type& prefix, size_t height,
+
+    void post(const std::vector<payment_address>& addresses, size_t height,
+        const hash_digest& block_hash, const transaction_type& tx);
+    void post(const std::vector<binary_type>& prefixes, size_t height,
         const hash_digest& block_hash, const transaction_type& tx);
+    void post(const payment_address& address, size_t height,
+        const hash_digest& block_hash, const transaction_type& tx);
+    void post(const binary_type& prefix, size_t height,
+        const hash_digest& block_hash, const transaction_type& tx);
+
     void sweep_expired();
 
     sequencer strand_;

install.sh

@@ -61,11 +61,13 @@ set -e
 #------------------------------------------------------------------------------
 SEQUENTIAL=1
 OS=`uname -s`
-if [[ $TRAVIS == true ]]; then
+if [[ $PARALLEL ]]; then
+    echo "Using shell-defined PARALLEL value."
+elif [[ $TRAVIS == true ]]; then
     PARALLEL=$SEQUENTIAL
 elif [[ $OS == Linux ]]; then
     PARALLEL=`nproc`
-elif [[ $OS == Darwin ]]; then
+elif [[ ($OS == Darwin) || ($OS == OpenBSD) ]]; then
     PARALLEL=`sysctl -n hw.ncpu`
 else
     echo "Unsupported system: $OS"
@@ -81,14 +83,20 @@ if [[ $OS == Darwin ]]; then
     export CC="clang"
     export CXX="clang++"
     LIBC="libc++"
-    
+
     # Always initialize prefix on OSX so default is useful.
     PREFIX="/usr/local"
+elif [[ $OS == OpenBSD ]]; then
+    make() { gmake "$@"; }
+    export CC="egcc"
+    export CXX="eg++"
+    LIBC="libestdc++"
 else
     LIBC="libstdc++"
 fi
 echo "Make with cc: $CC"
 echo "Make with cxx: $CXX"
+echo "Make with stdlib: $LIBC"
 
 # Define compiler specific settings.
 #------------------------------------------------------------------------------
@@ -357,7 +365,7 @@ initialize_boost_icu()
 
 initialize_icu_packages()
 {
-    if [[ $OS == Darwin && !($BUILD_ICU) ]]; then
+    if [[ ($OS == Darwin) && !($BUILD_ICU) ]]; then
         # Update PKG_CONFIG_PATH for ICU package installations on OSX.
         # OSX provides libicucore.dylib with no pkgconfig and doesn't support
         # renaming or important features, so we can't use that.

src/dispatch.cpp

@@ -268,9 +268,13 @@ static console_result run(const settings_type& config, std::ostream& output,
         }
 
     // By the current design this must be kept in scope until end of run.
-    subscribe_manager subscriber(full_node);
+    subscribe_manager subscriber(full_node, config.server.subscription_limit,
+        config.server.subscription_expiration_minutes);
+
+    request_worker worker(config.server.log_requests,
+        config.server.heartbeat_interval_seconds,
+        config.server.polling_interval_milliseconds);
 
-    request_worker worker;
     if (config.server.queries_enabled)
     {
         if (!worker.start(config))

src/server_node.cpp

@@ -36,6 +36,8 @@ using namespace bc::chain;
 using namespace bc::node;
 using std::placeholders::_1;
 using std::placeholders::_2;
+using std::placeholders::_3;
+using std::placeholders::_4;
 
 const settings_type server_node::defaults
 {
@@ -111,7 +113,20 @@ server_node::server_node(const settings_type& config)
 
 bool server_node::start(const settings_type& config)
 {
-    return full_node::start(config);
+    if (!full_node::start(config))
+        return false;
+
+    // Subscribe to reorganizations.
+    blockchain_.subscribe_reorganize(
+        std::bind(&server_node::handle_reorg,
+            this, _1, _2, _3, _4));
+
+    // Subscribe to mempool acceptances.
+    tx_pool_.subscribe_transaction(
+        std::bind(&server_node::handle_tx,
+            this, _1, _2, _3));
+
+    return true;
 }
 
 void server_node::subscribe_blocks(block_notify_callback notify_block)
@@ -124,38 +139,51 @@ void server_node::subscribe_transactions(transaction_notify_callback notify_tx)
     tx_subscriptions_.push_back(notify_tx);
 }
 
-void server_node::new_unconfirm_valid_tx(const std::error_code& ec,
+bool server_node::handle_tx(const std::error_code& ec,
     const index_list& unconfirmed, const transaction_type& tx)
 {
-    full_node::new_unconfirm_valid_tx(ec, unconfirmed, tx);
-
     if (ec == bc::error::service_stopped)
-        return;
+        return false;
+
+    if (ec)
+    {
+        log_error(LOG_SERVICE)
+            << "Failure handling tx: " << ec.message();
+        return false;
+    }
 
     // Fire server protocol tx subscription notifications.
     for (const auto notify: tx_subscriptions_)
         notify(tx);
+
+    return true;
 }
 
-void server_node::broadcast_new_blocks(const std::error_code& ec,
-    uint32_t fork_point, const blockchain::block_list& new_blocks,
-    const blockchain::block_list& replaced_blocks)
+bool server_node::handle_reorg(const std::error_code& ec, uint64_t fork_point,
+    const blockchain::block_list& new_blocks, const blockchain::block_list&)
 {
-    broadcast_new_blocks(ec, fork_point, new_blocks, replaced_blocks);
-
     if (ec == bc::error::service_stopped)
-        return;
+        return false;
 
     if (fork_point < minimum_start_height_)
-        return;
+        return false;
+
+    if (ec)
+    {
+        log_error(LOG_SERVICE)
+            << "Failure handling reorg: " << ec.message();
+        return false;
+    }
+
+    BITCOIN_ASSERT(fork_point < max_size_t - new_blocks.size());
+    auto height = static_cast<size_t>(fork_point);
 
     // Fire server protocol block subscription notifications.
     for (auto new_block: new_blocks)
-    {
-        const size_t height = ++fork_point;
         for (const auto notify: block_sunscriptions_)
-            notify(height, *new_block);
-    }
+            notify(++height, *new_block);
+
+    return true;
 }
 
 void server_node::fullnode_fetch_history(server_node& node,