Draft vision of market events model

libraries/app/database_api.cpp

@@ -118,6 +118,9 @@ class database_api_impl : public std::enable_shared_from_this<database_api_impl>
       void subscribe_to_market(std::function<void(const variant&)> callback, const std::string& a, const std::string& b);
       void unsubscribe_from_market(const std::string& a, const std::string& b);
 
+      void subscribe_to_market_events(std::function<void(const variant&)> callback, const std::string& a, const std::string& b);
+      void parse_market_operations();
+
       market_ticker                      get_ticker( const string& base, const string& quote, bool skip_order_book = false )const;
       market_volume                      get_24_volume( const string& base, const string& quote )const;
       order_book                         get_order_book( const string& base, const string& quote, unsigned limit = 50 )const;
@@ -333,6 +336,7 @@ class database_api_impl : public std::enable_shared_from_this<database_api_impl>
       boost::signals2::scoped_connection                                                                                           _applied_block_connection;
       boost::signals2::scoped_connection                                                                                           _pending_trx_connection;
       map< pair<asset_id_type,asset_id_type>, std::function<void(const variant&)> >      _market_subscriptions;
+      map< pair<asset_id_type,asset_id_type>, std::function<void(const variant&)> >      _market_events_subscriptions;
       graphene::chain::database&                                                                                                            _db;
       const application_options* _app_options = nullptr;
 };
@@ -1396,6 +1400,21 @@ void database_api_impl::subscribe_to_market(std::function<void(const variant&)>
    _market_subscriptions[ std::make_pair(asset_a_id,asset_b_id) ] = callback;
 }
 
+void database_api::subscribe_to_market_events(std::function<void(const variant&)> callback, const std::string& a, const std::string& b)
+{
+   my->subscribe_to_market_events( callback, a, b );
+}
+
+void database_api_impl::subscribe_to_market_events(std::function<void(const variant&)> callback, const std::string& a, const std::string& b)
+{
+   auto asset_a_id = get_asset_from_string(a)->id;
+   auto asset_b_id = get_asset_from_string(b)->id;
+
+   if(asset_a_id > asset_b_id) std::swap(asset_a_id,asset_b_id);
+   FC_ASSERT(asset_a_id != asset_b_id);
+   _market_events_subscriptions[ std::make_pair(asset_a_id,asset_b_id) ] = callback;
+}
+
 void database_api::unsubscribe_from_market(const std::string& a, const std::string& b)
 {
    my->unsubscribe_from_market( a, b );
@@ -2491,6 +2510,61 @@ void database_api_impl::handle_object_changed(bool force_notify, bool full_objec
    }
 }
 
+void database_api_impl::parse_market_operations()
+{
+   if (_market_events_subscriptions.size() == 0)
+   {
+      return;
+   }
+
+   const auto& ops = _db.get_applied_operations();
+   map< std::pair<asset_id_type,asset_id_type>, vector<pair<operation, operation_result>> > markets_ops;
+
+   for(const optional< operation_history_object >& o_op : ops)
+   {
+      if( !o_op.valid() )
+      {
+         continue;
+      }
+      const operation_history_object& op = *o_op;
+
+      optional< std::pair<asset_id_type,asset_id_type> > market;
+      switch(op.op.which())
+      {
+         case operation::tag<limit_order_create_operation>::value:
+            market = op.op.get<limit_order_create_operation>().get_market();
+            break;
+
+         case operation::tag<fill_order_operation>::value:
+            market = op.op.get<fill_order_operation>().get_market();
+            break;
+
+         case operation::tag<limit_order_cancel_operation>::value:
+            market = op.op.get<limit_order_cancel_operation>().get_market();
+            break;
+         default: break;
+      }
+
+      if( market.valid() && _market_events_subscriptions.count(*market) )
+      {
+         markets_ops[*market].emplace_back(std::make_pair(op.op, op.result));
+      }
+   }
+   /// we need to ensure the database_api is not deleted for the life of the async operation
+   auto capture_this = shared_from_this();
+   fc::async([this, capture_this, markets_ops]() {
+      for(auto item : markets_ops)
+      {
+         auto itr = _market_events_subscriptions.find(item.first);
+         if( itr != _market_events_subscriptions.end() )
+         {
+            itr->second(fc::variant(item.second, GRAPHENE_NET_MAX_NESTED_OBJECTS));
+         }
+      }
+   });
+}
+
+
 /** note: this method cannot yield because it is called in the middle of
  * apply a block.
  */
@@ -2505,6 +2579,8 @@ void database_api_impl::on_applied_block()
       });
    }
 
+   parse_market_operations();
+
    if(_market_subscriptions.size() == 0)
       return;
 

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

@@ -484,6 +484,9 @@ class database_api
        */
       void unsubscribe_from_market( const std::string& a, const std::string& b );
 
+
+      void subscribe_to_market_events(std::function<void(const variant&)> callback, const std::string& a, const std::string& b);
+
       /**
        * @brief Returns the ticker for the market assetA:assetB
        * @param a String name of the first asset

libraries/chain/db_maint.cpp

@@ -737,9 +737,11 @@ void create_buyback_orders( database& db )
 
             limit_order_id_type order_id = db.apply_operation( buyback_context, create_vop ).get< object_id_type >();
 
-            if( db.find( order_id ) != nullptr )
+            const auto order = db.find( order_id );
+            if( order != nullptr )
             {
                limit_order_cancel_operation cancel_vop;
+               cancel_vop.market = order->get_market();
                cancel_vop.fee = asset( 0, asset_id_type() );
                cancel_vop.order = order_id;
                cancel_vop.fee_paying_account = buyback_account.id;

libraries/chain/db_market.cpp

@@ -215,6 +215,7 @@ void database::cancel_limit_order( const limit_order_object& order, bool create_
    asset deferred_paid_fee = order.deferred_paid_fee;
    if( create_virtual_op )
    {
+      vop.market = order.get_market();
       vop.order = order.id;
       vop.fee_paying_account = order.seller;
       // only deduct fee if not skipping fee, and there is any fee deferred

libraries/chain/include/graphene/chain/protocol/market.hpp

@@ -90,6 +90,12 @@ namespace graphene { namespace chain {
       /** must be order->seller */
       account_id_type     fee_paying_account;
       extensions_type   extensions;
+      pair<asset_id_type,asset_id_type> market;
+
+      pair<asset_id_type,asset_id_type> get_market()const
+      {
+         return market;
+      }
 
       account_id_type fee_payer()const { return fee_paying_account; }
       void            validate()const;

libraries/wallet/wallet.cpp

@@ -2140,6 +2140,8 @@ class wallet_api_impl
          signed_transaction trx;
 
          limit_order_cancel_operation op;
+         const auto order = get_object<limit_order_object>(order_id);
+         op.market = order.get_market();
          op.fee_paying_account = get_object<limit_order_object>(order_id).seller;
          op.order = order_id;
          trx.operations = {op};

tests/common/database_fixture.cpp

@@ -763,6 +763,7 @@ const limit_order_object* database_fixture::create_sell_order( const account_obj
 asset database_fixture::cancel_limit_order( const limit_order_object& order )
 {
   limit_order_cancel_operation cancel_order;
+  cancel_order.market = order.get_market();
   cancel_order.fee_paying_account = order.seller;
   cancel_order.order = order.id;
   trx.operations.push_back(cancel_order);

tests/common/database_fixture.hpp

@@ -155,7 +155,7 @@ extern uint32_t GRAPHENE_TESTING_GENESIS_TIMESTAMP;
 
 #define ACTOR(name) \
    PREP_ACTOR(name) \
-   const auto& name = create_account(BOOST_PP_STRINGIZE(name), name ## _public_key); \
+   const auto name = create_account(BOOST_PP_STRINGIZE(name), name ## _public_key); \
    graphene::chain::account_id_type name ## _id = name.id; (void)name ## _id;
 
 #define GET_ACTOR(name) \

tests/tests/database_api_tests.cpp

@@ -25,6 +25,7 @@
 #include <boost/test/unit_test.hpp>
 
 #include <graphene/app/database_api.hpp>
+#include <graphene/chain/hardfork.hpp>
 
 #include <fc/crypto/digest.hpp>
 
@@ -941,4 +942,118 @@ BOOST_AUTO_TEST_CASE( verify_authority_multiple_accounts )
    }
 }
 
+BOOST_AUTO_TEST_CASE( subscribe_to_market_test )
+{
+   try {
+      ACTORS( (alice) (bob) );
+
+      generate_blocks(HARDFORK_CORE_625_TIME);
+      set_expiration(db, trx);
+
+      const auto usd = create_user_issued_asset("USD");
+      const auto eur = create_user_issued_asset("EUR");
+      const auto core = asset_id_type()(db);
+
+      fund(alice);
+      issue_uia(bob_id, usd.amount(30000));
+      issue_uia(bob_id, eur.amount(30000));
+
+      graphene::app::database_api db_api(db);
+
+      generate_block();
+
+      auto market_event_count = 0u;
+      auto on_market_callback = [&]( const variant& v )
+      {
+         ++market_event_count;
+         //std::cout << "\n\non_market_callback: BEGIN " << market_event_count << std::endl;
+         //std::cout << fc::json::to_pretty_string(v) << std::endl;
+         //std::cout << "on_market_callback: END " << market_event_count << std::endl;
+      };
+
+      db_api.subscribe_to_market_events(on_market_callback, std::string(static_cast<object_id_type>(usd.get_id())),
+                                                            std::string(static_cast<object_id_type>(core.get_id())));
+
+      generate_block();
+
+      // Full match in the same block
+      {
+         create_sell_order(alice, core.amount(100), usd.amount(100));
+         create_sell_order(bob, usd.amount(100), core.amount(100));
+         generate_block();
+         fc::usleep( fc::seconds(2) );
+         // 4 Events were got at once:
+         // [limit_order_create_operation, limit_order_create_operation, fill_order_operation, fill_order_operation]
+         BOOST_CHECK_EQUAL(market_event_count, 1);
+         market_event_count = 0u;
+      }
+      // Full match in different blocks
+       {
+          generate_block();
+          create_sell_order(alice, core.amount(100), usd.amount(100));
+          generate_block();
+          generate_block();
+          create_sell_order(bob, usd.amount(100), core.amount(100));
+          generate_block();
+          fc::usleep( fc::seconds(2) );
+         // 4 Events were got in two call of callback:
+         // 1: [limit_order_create_operation]
+         // 2: [limit_order_create_operation, fill_order_operation, fill_order_operation]
+
+          BOOST_CHECK_EQUAL(market_event_count, 2);
+          market_event_count = 0u;
+      }
+      set_expiration(db, trx);
+      // Partly match in the same block and cancel second order
+      {
+          generate_block();
+          create_sell_order(alice, core.amount(100), usd.amount(100));
+          auto order = *create_sell_order(bob, usd.amount(200), core.amount(200));
+          generate_block();
+          cancel_limit_order(order);
+          generate_block();
+          fc::usleep( fc::seconds(2) );
+
+         // 4 Events were got in two call of callback:
+         // 1: [limit_order_create_operation, limit_order_create_operation, fill_order_operation, fill_order_operation]
+         // 2: [limit_order_cancel_operation]
+          BOOST_CHECK_EQUAL(market_event_count, 2);
+          market_event_count = 0u;
+      }
+
+      // Partly match in the same block and cancel second order by expiretion
+      {
+         generate_block();
+         create_sell_order(alice, core.amount(100), usd.amount(100), db.head_block_time() + fc::seconds(10));
+         create_sell_order(bob, usd.amount(200), core.amount(200), db.head_block_time() + fc::seconds(10));
+         generate_block();
+         generate_block();
+         fc::usleep( fc::seconds(2) );
+         // 4 Events were got in two call of callback:
+         // 1: [limit_order_create_operation, limit_order_create_operation, fill_order_operation, fill_order_operation]
+         // 2: [limit_order_cancel_operation]
+          BOOST_CHECK_EQUAL(market_event_count, 2);
+          market_event_count = 0u;
+      }
+      set_expiration(db, trx);
+      // Partly match in the different blocks and cancel second order by expiretion
+      {
+         generate_block();
+         create_sell_order(alice, core.amount(100), usd.amount(100), db.head_block_time() + fc::seconds(10));
+         generate_block();
+         create_sell_order(bob, usd.amount(200), core.amount(200), db.head_block_time() + fc::seconds(10));
+         generate_block();
+         generate_block();
+         fc::usleep( fc::seconds(2) );
+         // 4 Events were got in three call of callback:
+         // 1: [limit_order_create_operation]
+         // 2: [limit_order_create_operation, fill_order_operation, fill_order_operation]
+         // 3: [limit_order_cancel_operation]
+          BOOST_CHECK_EQUAL(market_event_count, 3);
+          market_event_count = 0u;
+      }
+
+   } FC_LOG_AND_RETHROW()
+}
+
 BOOST_AUTO_TEST_SUITE_END()