spark/src/lib/orderman/actors/ServiceBridge.cpp

#include "ServiceBridge.h"

#include "lib/api/public/ServiceContext.h"
#include "lib/mercury/broker/BrokerProxy.h"
#include "lib/orderman/events/ActivateOrderEvent.h"
#include "lib/orderman/events/CancelOrderEvent.h"
#include "lib/orderman/events/CrackOrderEvent.h"
#include "lib/orderman/events/NewAdminDataEvent.h"
#include "lib/orderman/events/NewSecuritiesEvent.h"
#include "lib/orderman/events/OrderManagerReadyEvent.h"
#include "lib/orderman/events/StoredFillEvent.h"
#include "lib/orderman/events/StoredOrderEvent.h"
#include "lib/orderman/events/UpdateOrderEvent.h"
#include "lib/orderman/order_factory/OrderFactory.h"
#include "lib/orderman/order_router/OrderRouter.h"
#include "lib/orderman/securities/MapSecurityDB.h"
#include "lib/utils/PBUtils.h"

#include "lib/api/services/OrderService.pb.h"
#include "lib/orderman/orderman_utils/orderman_common.h"
#include "lib/orderman/orders/Order.h"

namespace bts {
namespace orderman {

struct ServiceBridge::PImpl {
   PImpl(OnReady const &on_ready, OnOrders const &on_orders,
         OnFills const &on_fills, DynamicOrderRegistry const &dyn_order_registry, mercury::BrokerProxy &service_call_marshaller)
       : on_ready(on_ready), on_orders(on_orders), on_fills(on_fills),
         have_orders(false), have_securities(false), order_manager_ready(false),
         sec_db(), order_router(),
         order_factory(sec_db, order_router, dyn_order_registry),
         known_orders(), service_call_marshaller(service_call_marshaller) {}

   OnReady on_ready;
   OnOrders on_orders;
   OnFills on_fills;
   bool have_orders;
   bool have_securities;
   bool order_manager_ready;
   securities::MapSecurityDB sec_db;
   OrderRouter order_router;
   OrderFactory order_factory;
   mercury::Stream<ActivateOrderEvent> activate_order_pub;
   mercury::Stream<UpdateOrderEvent> update_order_pub;
   mercury::Stream<CancelOrderEvent> cancel_order_pub;
   mercury::Stream<CrackOrderEvent> crack_order_pub;
   mercury::Stream<NewAdminDataEvent> new_admin_data_pub;
   std::unordered_set<OrderKey> known_orders;
   std::vector<std::shared_ptr<const api::admin::AdminData>> admin_data_updates;
   mercury::BrokerProxy &service_call_marshaller;
   std::promise<void> ready;
};

ServiceBridge::ServiceBridge(OnReady const &on_ready, OnOrders const &on_orders,
                             OnFills const &on_fills,
                             DynamicOrderRegistry const &dyn_order_registry,
                             mercury::BrokerProxy &service_call_marshaller)							 
    : me(new PImpl(on_ready, on_orders, on_fills, dyn_order_registry, service_call_marshaller)) {}

ServiceBridge::~ServiceBridge() {}

void ServiceBridge::block_until_ready() { me->ready.get_future().wait(); }

void ServiceBridge::init() {
   subscribe<NewAdminDataEvent>([this](auto const &u) { this->on(u); });
   subscribe<NewSecuritiesEvent>([this](auto const &u) { this->on(u); });
   subscribe<StoredFillEvent>([this](auto const &u) { this->on(u); });
   subscribe<StoredOrderEvent>([this](auto const &u) { this->on(u); });
   subscribe<OrderManagerReadyEvent>([this](auto const &u) { this->on(u); });

   me->activate_order_pub = make_stream<ActivateOrderEvent>();
   me->update_order_pub = make_stream<UpdateOrderEvent>();
   me->cancel_order_pub = make_stream<CancelOrderEvent>();
   me->crack_order_pub = make_stream<CrackOrderEvent>();
   me->new_admin_data_pub = make_stream<NewAdminDataEvent>();
   me->ready.set_value();
}

void ServiceBridge::on(mercury::Update<StoredFillEvent> const &update) {
   if (update.events) {
      std::vector<api::orderman::Fill> fills;
      fills.reserve(update.events->size());
      for (auto ev : *(update.events))
         fills.push_back(ev.fill);
      me->on_fills(fills);
   }
}

void ServiceBridge::on(mercury::Update<StoredOrderEvent> const &update) {
   if (update.events) {
      bool was_ready = is_ready();
      me->have_orders = true;

      std::vector<api::orderman::OrderRecord> orders;
      orders.reserve(update.events->size());
      for (auto ev : *(update.events)) {
         me->known_orders.insert(OrderKey(ev.order_record.key()));
         orders.push_back(ev.order_record);
      }
      me->on_orders(orders);
      maybe_signal_ready(was_ready);
   }
}

void ServiceBridge::on(mercury::Update<NewSecuritiesEvent> const &update) {
   if (update.events) {

      // update our security db
      for (auto ev : *(update.events)) {
         auto sec_ptr = ev.security;
         me->sec_db.securities[SecurityKey(sec_ptr->id())] = sec_ptr;
      }
      bool was_ready = is_ready();
      me->have_securities = true;
      maybe_signal_ready(was_ready);
   }
}

void ServiceBridge::on(mercury::Update<OrderManagerReadyEvent> const &update) {
   if (update.events) {
      bool was_ready = is_ready();
      me->order_manager_ready = true;
      maybe_signal_ready(was_ready);
   }
}

void ServiceBridge::on(mercury::Update<NewAdminDataEvent> const &update) {
   if (update.events)
      for (auto ev : *(update.events))
         me->order_router.update(*ev.admin_data);
}

api::orderman::Response
ServiceBridge::create_order(api::orderman::Create const &c) {
   LOG(OrderMan, Info, "Received new order from client: {}",
       utils::pb::to_string(c));

   std::promise<api::orderman::Response> p;
   auto f = p.get_future();

   me->service_call_marshaller.schedule_action([this, &c, &p]() {
      api::orderman::Response r;

      if (is_ready()) {
         auto &order =
             me->order_factory.create(null_order_key(), c.new_order());
         me->known_orders.insert(order.key());
         publish(me->activate_order_pub, order, Price(c.price()),
                 Quantity(c.quantity()), PerfTrace());
         r.set_order_key(order.key().value());
         r.set_error_code(api::orderman::ORDER_ERROR_OK);
      } else
         r.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      p.set_value(r);
   });
   return f.get();
}

api::orderman::Response
ServiceBridge::update_order(api::orderman::Update const &c) {
   LOG(OrderMan, Info, "Received order update from client: {}",
       utils::pb::to_string(c));

   std::promise<api::orderman::Response> p;
   auto f = p.get_future();

   me->service_call_marshaller.schedule_action([this, &c, &p]() {
      api::orderman::Response r;

      if (is_ready()) {
         auto ok = OrderKey(c.order_key());
         r.set_order_key(c.order_key());
         if (me->known_orders.count(ok)) {
            publish(me->update_order_pub, ok, Price(c.price()),
                    Quantity(c.quantity()));
            r.set_error_code(api::orderman::ORDER_ERROR_OK);
         } else
            r.set_error_code(api::orderman::ORDER_ERROR_UNKNOWN_ORDER);
      } else
         r.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      p.set_value(r);
   });
   return f.get();
}

api::orderman::Response
ServiceBridge::cancel_order(api::orderman::Cancel const &c) {
   LOG(OrderMan, Info, "Received order cancel from client: {}",
       utils::pb::to_string(c));

   std::promise<api::orderman::Response> p;
   auto f = p.get_future();

   me->service_call_marshaller.schedule_action([this, &c, &p]() {
      api::orderman::Response r;

      if (is_ready()) {
         auto ok = OrderKey(c.order_key());
         r.set_order_key(c.order_key());
         if (me->known_orders.count(ok)) {
            publish(me->cancel_order_pub, ok);
            r.set_error_code(api::orderman::ORDER_ERROR_OK);
         } else
            r.set_error_code(api::orderman::ORDER_ERROR_UNKNOWN_ORDER);
      } else
         r.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      p.set_value(r);
   });
   return f.get();
}

api::orderman::Response
ServiceBridge::crack_order(api::orderman::Crack const &c) {
   std::promise<api::orderman::Response> p;
   auto f = p.get_future();

   me->service_call_marshaller.schedule_action([this, &c, &p]() {
      api::orderman::Response r;

      if (is_ready()) {
         auto ok = OrderKey(c.order_key());
         r.set_order_key(c.order_key());
         if (me->known_orders.count(ok)) {
            publish(me->crack_order_pub, ok);
            r.set_error_code(api::orderman::ORDER_ERROR_OK);
         } else
            r.set_error_code(api::orderman::ORDER_ERROR_UNKNOWN_ORDER);
      } else
         r.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
      p.set_value(r);
   });
   return f.get();
}

void ServiceBridge::update_admin_data(
    std::shared_ptr<const api::admin::AdminData> const &admin) {
   me->service_call_marshaller.schedule_action([this, admin]() {
      me->admin_data_updates.push_back(admin);
      publish(me->new_admin_data_pub, *admin);
   });
}

bool ServiceBridge::is_ready() const {
   return me->have_orders && me->have_securities && me->order_manager_ready;
}

void ServiceBridge::maybe_signal_ready(bool old_ready) {
   if (!old_ready && is_ready())
      me->on_ready();
}
} // namespace orderman
} // namespace bts