Merge pull request #725 from bitshares/abitmore-patch-2

Add application option to disable `notify_remove_create` parameter in `set_subscribe_callback` API

libraries/app/api.cpp

@@ -78,7 +78,7 @@ namespace graphene { namespace app {
     {
        if( api_name == "database_api" )
        {
-          _database_api = std::make_shared< database_api >( std::ref( *_app.chain_database() ) );
+          _database_api = std::make_shared< database_api >( std::ref( *_app.chain_database() ), &( _app.get_options() ) );
        }
        else if( api_name == "block_api" )
        {

libraries/app/application.cpp

@@ -111,6 +111,7 @@ namespace detail {
       fc::optional<fc::temp_file> _lock_file;
       bool _is_block_producer = false;
       bool _force_validate = false;
+      application_options _app_options;
 
       void reset_p2p_node(const fc::path& data_dir)
       { try {
@@ -225,7 +226,9 @@ namespace detail {
                std::string hostname = endpoint_string.substr(0, colon_pos);
                std::vector<fc::ip::endpoint> endpoints = fc::resolve(hostname, port);
                if (endpoints.empty())
-                  FC_THROW_EXCEPTION(fc::unknown_host_exception, "The host name can not be resolved: ${hostname}", ("hostname", hostname));
+                  FC_THROW_EXCEPTION( fc::unknown_host_exception,
+                                      "The host name can not be resolved: ${hostname}",
+                                      ("hostname", hostname) );
                return endpoints;
             }
             catch (const boost::bad_lexical_cast&)
@@ -334,10 +337,14 @@ namespace detail {
                bool modified_genesis = false;
                if( _options->count("genesis-timestamp") )
                {
-                  genesis.initial_timestamp = fc::time_point_sec( fc::time_point::now() ) + genesis.initial_parameters.block_interval + _options->at("genesis-timestamp").as<uint32_t>();
-                  genesis.initial_timestamp -= genesis.initial_timestamp.sec_since_epoch() % genesis.initial_parameters.block_interval;
+                  genesis.initial_timestamp = fc::time_point_sec( fc::time_point::now() )
+                                            + genesis.initial_parameters.block_interval
+                                            + _options->at("genesis-timestamp").as<uint32_t>();
+                  genesis.initial_timestamp -= ( genesis.initial_timestamp.sec_since_epoch()
+                                                 % genesis.initial_parameters.block_interval );
                   modified_genesis = true;
-                  std::cerr << "Used genesis timestamp:  " << genesis.initial_timestamp.to_iso_string() << " (PLEASE RECORD THIS)\n";
+                  std::cerr << "Used genesis timestamp:  " << genesis.initial_timestamp.to_iso_string()
+                            << " (PLEASE RECORD THIS)\n";
                }
                if( _options->count("dbg-init-key") )
                {
@@ -404,6 +411,12 @@ namespace detail {
             _force_validate = true;
          }
 
+         if( _options->count("enable-subscribe-to-all") )
+            _app_options.enable_subscribe_to_all = _options->at("enable-subscribe-to-all").as<bool>();
+
+         if( _active_plugins.find( "market_history" ) != _active_plugins.end() )
+            _app_options.has_market_history_plugin = true;
+
          if( _options->count("api-access") ) {
 
             if(fc::exists(_options->at("api-access").as<boost::filesystem::path>()))
@@ -419,7 +432,6 @@ namespace detail {
                std::exit(EXIT_FAILURE);
             }
          }
-
          else
          {
             // TODO:  Remove this generous default access policy
@@ -506,7 +518,9 @@ namespace detail {
             // you can help the network code out by throwing a block_older_than_undo_history exception.
             // when the net code sees that, it will stop trying to push blocks from that chain, but
             // leave that peer connected so that they can get sync blocks from us
-            bool result = _chain_db->push_block(blk_msg.block, (_is_block_producer | _force_validate) ? database::skip_nothing : database::skip_transaction_signatures);
+            bool result = _chain_db->push_block( blk_msg.block,
+                                                 (_is_block_producer | _force_validate) ?
+                                                    database::skip_nothing : database::skip_transaction_signatures );
 
             // the block was accepted, so we now know all of the transactions contained in the block
             if (!sync_mode)
@@ -527,7 +541,9 @@ namespace detail {
          } catch ( const graphene::chain::unlinkable_block_exception& e ) {
             // translate to a graphene::net exception
             elog("Error when pushing block:\n${e}", ("e", e.to_detail_string()));
-            FC_THROW_EXCEPTION(graphene::net::unlinkable_block_exception, "Error when pushing block:\n${e}", ("e", e.to_detail_string()));
+            FC_THROW_EXCEPTION( graphene::net::unlinkable_block_exception,
+                                "Error when pushing block:\n${e}",
+                                ("e", e.to_detail_string()) );
          } catch( const fc::exception& e ) {
             elog("Error when pushing block:\n${e}", ("e", e.to_detail_string()));
             throw;
@@ -610,7 +626,8 @@ namespace detail {
                break;
              }
            if (!found_a_block_in_synopsis)
-             FC_THROW_EXCEPTION(graphene::net::peer_is_on_an_unreachable_fork, "Unable to provide a list of blocks starting at any of the blocks in peer's synopsis");
+             FC_THROW_EXCEPTION( graphene::net::peer_is_on_an_unreachable_fork,
+                                 "Unable to provide a list of blocks starting at any of the blocks in peer's synopsis" );
          }
          for( uint32_t num = block_header::num_from_id(last_known_block_id);
               num <= _chain_db->head_block_num() && result.size() < limit;
@@ -775,7 +792,8 @@ namespace detail {
               {
                 // unable to get fork history for some reason.  maybe not linked?
                 // we can't return a synopsis of its chain
-                elog("Unable to construct a blockchain synopsis for reference hash ${hash}: ${exception}", ("hash", reference_point)("exception", e));
+                elog( "Unable to construct a blockchain synopsis for reference hash ${hash}: ${exception}",
+                      ("hash", reference_point)("exception", e) );
                 throw;
               }
               if (non_fork_high_block_num < low_block_num)
@@ -922,17 +940,24 @@ void application::set_program_options(boost::program_options::options_descriptio
 {
    configuration_file_options.add_options()
          ("p2p-endpoint", bpo::value<string>(), "Endpoint for P2P node to listen on")
-         ("seed-node,s", bpo::value<vector<string>>()->composing(), "P2P nodes to connect to on startup (may specify multiple times)")
-         ("seed-nodes", bpo::value<string>()->composing(), "JSON array of P2P nodes to connect to on startup")
-         ("checkpoint,c", bpo::value<vector<string>>()->composing(), "Pairs of [BLOCK_NUM,BLOCK_ID] that should be enforced as checkpoints.")
-         ("rpc-endpoint", bpo::value<string>()->implicit_value("127.0.0.1:8090"), "Endpoint for websocket RPC to listen on")
-         ("rpc-tls-endpoint", bpo::value<string>()->implicit_value("127.0.0.1:8089"), "Endpoint for TLS websocket RPC to listen on")
+         ("seed-node,s", bpo::value<vector<string>>()->composing(),
+          "P2P nodes to connect to on startup (may specify multiple times)")
+         ("seed-nodes", bpo::value<string>()->composing(),
+          "JSON array of P2P nodes to connect to on startup")
+         ("checkpoint,c", bpo::value<vector<string>>()->composing(),
+          "Pairs of [BLOCK_NUM,BLOCK_ID] that should be enforced as checkpoints.")
+         ("rpc-endpoint", bpo::value<string>()->implicit_value("127.0.0.1:8090"),
+          "Endpoint for websocket RPC to listen on")
+         ("rpc-tls-endpoint", bpo::value<string>()->implicit_value("127.0.0.1:8089"),
+          "Endpoint for TLS websocket RPC to listen on")
          ("server-pem,p", bpo::value<string>()->implicit_value("server.pem"), "The TLS certificate file for this server")
          ("server-pem-password,P", bpo::value<string>()->implicit_value(""), "Password for this certificate")
          ("genesis-json", bpo::value<boost::filesystem::path>(), "File to read Genesis State from")
          ("dbg-init-key", bpo::value<string>(), "Block signing key to use for init witnesses, overrides genesis file")
          ("api-access", bpo::value<boost::filesystem::path>(), "JSON file specifying API permissions")
          ("plugins", bpo::value<string>(), "Space-separated list of plugins to activate")
+         ("enable-subscribe-to-all", bpo::value<bool>()->implicit_value(false),
+          "Whether allow API clients to subscribe to universal object creation and removal events")
          ;
    command_line_options.add(configuration_file_options);
    command_line_options.add_options()
@@ -943,7 +968,8 @@ void application::set_program_options(boost::program_options::options_descriptio
          ("replay-blockchain", "Rebuild object graph by replaying all blocks")
          ("resync-blockchain", "Delete all blocks and re-sync with network from scratch")
          ("force-validate", "Force validation of all transactions")
-         ("genesis-timestamp", bpo::value<uint32_t>(), "Replace timestamp from genesis.json with current time plus this many seconds (experts only!)")
+         ("genesis-timestamp", bpo::value<uint32_t>(),
+          "Replace timestamp from genesis.json with current time plus this many seconds (experts only!)")
          ;
    command_line_options.add(_cli_options);
    configuration_file_options.add(_cfg_options);
@@ -1098,5 +1124,10 @@ void application::startup_plugins()
    return;
 }
 
+const application_options& application::get_options()
+{
+   return my->_app_options;
+}
+
 // namespace detail
 } }

libraries/app/database_api.cpp

@@ -53,7 +53,7 @@ class database_api_impl;
 class database_api_impl : public std::enable_shared_from_this<database_api_impl>
 {
    public:
-      database_api_impl( graphene::chain::database& db );
+      database_api_impl( graphene::chain::database& db, const application_options* app_options );
       ~database_api_impl();
 
 
@@ -230,6 +230,7 @@ class database_api_impl : public std::enable_shared_from_this<database_api_impl>
       boost::signals2::scoped_connection                                                                                           _pending_trx_connection;
       map< pair<asset_id_type,asset_id_type>, std::function<void(const variant&)> >      _market_subscriptions;
       graphene::chain::database&                                                                                                            _db;
+      const application_options* _app_options = nullptr;
 };
 
 //////////////////////////////////////////////////////////////////////
@@ -238,12 +239,13 @@ class database_api_impl : public std::enable_shared_from_this<database_api_impl>
 //                                                                  //
 //////////////////////////////////////////////////////////////////////
 
-database_api::database_api( graphene::chain::database& db )
-   : my( new database_api_impl( db ) ) {}
+database_api::database_api( graphene::chain::database& db, const application_options* app_options )
+   : my( new database_api_impl( db, app_options ) ) {}
 
 database_api::~database_api() {}
 
-database_api_impl::database_api_impl( graphene::chain::database& db ):_db(db)
+database_api_impl::database_api_impl( graphene::chain::database& db, const application_options* app_options )
+:_db(db), _app_options(app_options)
 {
    wlog("creating database api ${x}", ("x",int64_t(this)) );
    _new_connection = _db.new_objects.connect([this](const vector<object_id_type>& ids, const flat_set<account_id_type>& impacted_accounts) {
@@ -316,6 +318,12 @@ void database_api::set_subscribe_callback( std::function<void(const variant&)> c
 
 void database_api_impl::set_subscribe_callback( std::function<void(const variant&)> cb, bool notify_remove_create )
 {
+   if( notify_remove_create )
+   {
+      FC_ASSERT( _app_options && _app_options->enable_subscribe_to_all,
+                 "Subscribing to universal object creation and removal is disallowed in this server." );
+   }
+
    _subscribe_callback = cb;
    _notify_remove_create = notify_remove_create;
    _subscribed_accounts.clear();
@@ -1166,20 +1174,22 @@ market_ticker database_api::get_ticker( const string& base, const string& quote
 
 market_ticker database_api_impl::get_ticker( const string& base, const string& quote, bool skip_order_book )const
 {
-    const auto assets = lookup_asset_symbols( {base, quote} );
-    FC_ASSERT( assets[0], "Invalid base asset symbol: ${s}", ("s",base) );
-    FC_ASSERT( assets[1], "Invalid quote asset symbol: ${s}", ("s",quote) );
+   FC_ASSERT( _app_options && _app_options->has_market_history_plugin, "Market history plugin is not enabled." );
 
-    const fc::time_point_sec now = fc::time_point::now();
+   const auto assets = lookup_asset_symbols( {base, quote} );
+   FC_ASSERT( assets[0], "Invalid base asset symbol: ${s}", ("s",base) );
+   FC_ASSERT( assets[1], "Invalid quote asset symbol: ${s}", ("s",quote) );
+
+   const fc::time_point_sec now = fc::time_point::now();
 
-    market_ticker result;
-    result.time = now;
-    result.base = base;
-    result.quote = quote;
-    result.latest = "0";
-    result.lowest_ask = "0";
-    result.highest_bid = "0";
-    result.percent_change = "0";
+   market_ticker result;
+   result.time = now;
+   result.base = base;
+   result.quote = quote;
+   result.latest = "0";
+   result.lowest_ask = "0";
+   result.highest_bid = "0";
+   result.percent_change = "0";
 
    auto base_id = assets[0]->id;
    auto quote_id = assets[1]->id;
@@ -1236,16 +1246,16 @@ market_volume database_api::get_24_volume( const string& base, const string& quo
 
 market_volume database_api_impl::get_24_volume( const string& base, const string& quote )const
 {
-    const auto& ticker = get_ticker( base, quote, true );
+   const auto& ticker = get_ticker( base, quote, true );
 
-    market_volume result;
-    result.time = ticker.time;
-    result.base = ticker.base;
-    result.quote = ticker.quote;
-    result.base_volume = ticker.base_volume;
-    result.quote_volume = ticker.quote_volume;
+   market_volume result;
+   result.time = ticker.time;
+   result.base = ticker.base;
+   result.quote = ticker.quote;
+   result.base_volume = ticker.base_volume;
+   result.quote_volume = ticker.quote_volume;
 
-    return result;
+   return result;
 }
 
 order_book database_api::get_order_book( const string& base, const string& quote, unsigned limit )const
@@ -1300,6 +1310,8 @@ vector<market_volume> database_api::get_top_markets(uint32_t limit)const
 
 vector<market_volume> database_api_impl::get_top_markets(uint32_t limit)const
 {
+   FC_ASSERT( _app_options && _app_options->has_market_history_plugin, "Market history plugin is not enabled." );
+
    FC_ASSERT( limit <= 100 );
 
    const auto& volume_idx = _db.get_index_type<graphene::market_history::market_ticker_index>().indices().get<by_volume>();
@@ -1339,6 +1351,8 @@ vector<market_trade> database_api_impl::get_trade_history( const string& base,
                                                            fc::time_point_sec stop,
                                                            unsigned limit )const
 {
+   FC_ASSERT( _app_options && _app_options->has_market_history_plugin, "Market history plugin is not enabled." );
+
    FC_ASSERT( limit <= 100 );
 
    auto assets = lookup_asset_symbols( {base, quote} );
@@ -1428,6 +1442,8 @@ vector<market_trade> database_api_impl::get_trade_history_by_sequence(
                                                            fc::time_point_sec stop,
                                                            unsigned limit )const
 {
+   FC_ASSERT( _app_options && _app_options->has_market_history_plugin, "Market history plugin is not enabled." );
+
    FC_ASSERT( limit <= 100 );
    FC_ASSERT( start >= 0 );
    int64_t start_seq = -start;
@@ -2211,7 +2227,8 @@ void database_api_impl::handle_object_changed(bool force_notify, bool full_objec
          }
       }
 
-      broadcast_updates(updates);
+      if( updates.size() )
+         broadcast_updates(updates);
    }
 
    if( _market_subscriptions.size() )
@@ -2230,7 +2247,8 @@ void database_api_impl::handle_object_changed(bool force_notify, bool full_objec
          }
       }
 
-      broadcast_market_updates(broadcast_queue);
+      if( broadcast_queue.size() )
+         broadcast_market_updates(broadcast_queue);
    }
 }
 

libraries/app/include/graphene/app/application.hpp

@@ -35,6 +35,13 @@ namespace graphene { namespace app {
 
    class abstract_plugin;
 
+   class application_options
+   {
+      public:
+         bool enable_subscribe_to_all = false;
+         bool has_market_history_plugin = false;
+   };
+
    class application
    {
       public:
@@ -89,6 +96,8 @@ namespace graphene { namespace app {
          /// Emitted when syncing finishes (is_finished_syncing will return true)
          boost::signals2::signal<void()> syncing_finished;
 
+         const application_options& get_options();
+
       private:
          void enable_plugin( const string& name );
          void add_available_plugin( std::shared_ptr<abstract_plugin> p );

libraries/app/include/graphene/app/database_api.hpp

@@ -122,7 +122,7 @@ struct market_trade
 class database_api
 {
    public:
-      database_api(graphene::chain::database& db);
+      database_api(graphene::chain::database& db, const application_options* app_options = nullptr );
       ~database_api();
 
       /////////////
@@ -142,8 +142,29 @@ class database_api
       // Subscriptions //
       ///////////////////
 
-      void set_subscribe_callback( std::function<void(const variant&)> cb, bool clear_filter );
-      void set_pending_transaction_callback( std::function<void(const variant&)> cb );
+      /**
+       * @brief Register a callback handle which then can be used to subscribe to object database changes
+       * @param cb The callback handle to register
+       * @param nofity_remove_create Whether subscribe to universal object creation and removal events.
+       *        If this is set to true, the API server will notify all newly created objects and ID of all
+       *        newly removed objects to the client, no matter whether client subscribed to the objects.
+       *        By default, API servers don't allow subscribing to universal events, which can be changed
+       *        on server startup.
+       */
+      void set_subscribe_callback( std::function<void(const variant&)> cb, bool notify_remove_create );
+      /**
+       * @brief Register a callback handle which will get notified when a transaction is pushed to database
+       * @param cb The callback handle to register
+       *
+       * Note: a transaction can be pushed to database and be popped from database several times while
+       *   processing, before and after included in a block. Everytime when a push is done, the client will
+       *   be notified.
+       */
+      void set_pending_transaction_callback( std::function<void(const variant& signed_transaction_object)> cb );
+      /**
+       * @brief Register a callback handle which will get notified when a block is pushed to database
+       * @param cb The callback handle to register
+       */
       void set_block_applied_callback( std::function<void(const variant& block_id)> cb );
       /**
        * @brief Stop receiving any notifications