Backport ElasticSearch plugins from BitShares

libraries/app/CMakeLists.txt

@@ -5,7 +5,6 @@ add_library( graphene_app
              api.cpp
              application.cpp
              database_api.cpp
-             impacted.cpp
              plugin.cpp
              config_util.cpp
              ${HEADERS}
@@ -14,7 +13,7 @@ add_library( graphene_app
 
 # need to link graphene_debug_witness because plugins aren't sufficiently isolated #246
 #target_link_libraries( graphene_app graphene_market_history graphene_account_history graphene_chain fc graphene_db graphene_net graphene_utilities graphene_debug_witness )
-target_link_libraries( graphene_app graphene_market_history graphene_account_history graphene_accounts_list graphene_affiliate_stats graphene_chain fc graphene_db graphene_net graphene_time graphene_utilities graphene_debug_witness graphene_bookie )
+target_link_libraries( graphene_app graphene_market_history graphene_account_history graphene_accounts_list graphene_affiliate_stats graphene_chain fc graphene_db graphene_net graphene_time graphene_utilities graphene_debug_witness graphene_bookie graphene_elasticsearch )
 target_include_directories( graphene_app
                             PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/include"
                             "${CMAKE_CURRENT_SOURCE_DIR}/../egenesis/include" )

libraries/app/api.cpp

@@ -26,7 +26,6 @@
 #include <graphene/app/api.hpp>
 #include <graphene/app/api_access.hpp>
 #include <graphene/app/application.hpp>
-#include <graphene/app/impacted.hpp>
 #include <graphene/chain/database.hpp>
 #include <graphene/chain/get_config.hpp>
 #include <graphene/utilities/key_conversion.hpp>
@@ -581,6 +580,18 @@ namespace graphene { namespace app {
               start = node.operation_id;
         } catch(...) { return result; }
 
+        if(_app.is_plugin_enabled("elasticsearch")) {
+           auto es = _app.get_plugin<elasticsearch::elasticsearch_plugin>("elasticsearch");
+           if(es.get()->get_running_mode() != elasticsearch::mode::only_save) {
+              if(!_app.elasticsearch_thread)
+                 _app.elasticsearch_thread= std::make_shared<fc::thread>("elasticsearch");
+
+              return _app.elasticsearch_thread->async([&es, &account, &stop, &limit, &start]() {
+                 return es->get_account_history(account, stop, limit, start);
+              }, "thread invoke for method " BOOST_PP_STRINGIZE(method_name)).wait();
+           }
+        }
+
         const auto& hist_idx = db.get_index_type<account_transaction_history_index>();
         const auto& by_op_idx = hist_idx.indices().get<by_op>();
         auto index_start = by_op_idx.begin();

libraries/app/application.cpp

@@ -891,7 +891,8 @@ namespace detail {
       std::shared_ptr<fc::http::websocket_server>      _websocket_server;
       std::shared_ptr<fc::http::websocket_tls_server>  _websocket_tls_server;
 
-      std::map<string, std::shared_ptr<abstract_plugin>> _plugins;
+      std::map<string, std::shared_ptr<abstract_plugin>> _active_plugins;
+      std::map<string, std::shared_ptr<abstract_plugin>> _available_plugins;
 
       bool _is_finished_syncing = false;
    };
@@ -933,6 +934,7 @@ void application::set_program_options(boost::program_options::options_descriptio
          ("enable-standby-votes-tracking", bpo::value<bool>()->implicit_value(true),
           "Whether to enable tracking of votes of standby witnesses and committee members. "
           "Set it to true to provide accurate data to API clients, set to false for slightly better performance.")
+         ("plugins", bpo::value<string>(), "Space-separated list of plugins to activate")
          ;
    command_line_options.add(configuration_file_options);
    command_line_options.add_options()
@@ -978,6 +980,32 @@ void application::initialize(const fc::path& data_dir, const boost::program_opti
 
       std::exit(EXIT_SUCCESS);
    }
+
+   std::vector<string> wanted;
+   if( options.count("plugins") )
+   {
+      boost::split(wanted, options.at("plugins").as<std::string>(), [](char c){return c == ' ';});
+   }
+   else
+   {
+      wanted.push_back("witness");
+      wanted.push_back("account_history");
+      wanted.push_back("market_history");
+   }
+   int es_ah_conflict_counter = 0;
+   for (auto& it : wanted)
+   {
+      if(it == "account_history")
+         ++es_ah_conflict_counter;
+      if(it == "elasticsearch")
+         ++es_ah_conflict_counter;
+
+      if(es_ah_conflict_counter > 1) {
+         elog("Can't start program with elasticsearch and account_history plugin at the same time");
+         std::exit(EXIT_FAILURE);
+      }
+      if (!it.empty()) enable_plugin(it);
+   }
 }
 
 void application::startup()
@@ -995,7 +1023,12 @@ void application::startup()
 
 std::shared_ptr<abstract_plugin> application::get_plugin(const string& name) const
 {
-   return my->_plugins[name];
+   return my->_active_plugins[name];
+}
+
+bool application::is_plugin_enabled(const string& name) const
+{
+   return !(my->_active_plugins.find(name) == my->_active_plugins.end());
 }
 
 net::node_ptr application::p2p_node()
@@ -1028,14 +1061,21 @@ bool application::is_finished_syncing() const
    return my->_is_finished_syncing;
 }
 
-void graphene::app::application::add_plugin(const string& name, std::shared_ptr<graphene::app::abstract_plugin> p)
+void graphene::app::application::enable_plugin(const string& name)
+{
+   FC_ASSERT(my->_available_plugins[name], "Unknown plugin '" + name + "'");
+   my->_active_plugins[name] = my->_available_plugins[name];
+   my->_active_plugins[name]->plugin_set_app(this);
+}
+
+void graphene::app::application::add_available_plugin(std::shared_ptr<graphene::app::abstract_plugin> p)
 {
-   my->_plugins[name] = p;
+   my->_available_plugins[p->plugin_name()] = p;
 }
 
 void application::shutdown_plugins()
 {
-   for( auto& entry : my->_plugins )
+   for( auto& entry : my->_active_plugins )
       entry.second->plugin_shutdown();
    return;
 }
@@ -1049,14 +1089,14 @@ void application::shutdown()
 
 void application::initialize_plugins( const boost::program_options::variables_map& options )
 {
-   for( auto& entry : my->_plugins )
+   for( auto& entry : my->_active_plugins )
       entry.second->plugin_initialize( options );
    return;
 }
 
 void application::startup_plugins()
 {
-   for( auto& entry : my->_plugins )
+   for( auto& entry : my->_active_plugins )
       entry.second->plugin_startup();
    return;
 }