101 lines
3.1 KiB
C++
101 lines
3.1 KiB
C++
#include "MDClient.h"
|
|
#include "get_symbols.h"
|
|
#include "SecurityCache.h"
|
|
#include "SecurityOracle.h"
|
|
#include "lib/utils/TimeStamp.h"
|
|
#include "MBOProcessor.h"
|
|
#include <set>
|
|
|
|
namespace bts {
|
|
namespace md {
|
|
|
|
struct MDClientImpl{
|
|
utils::Dispatcher* _dispatcher;
|
|
api::ISyntheticSecurityService* _ss;
|
|
SecurityCache _cache;
|
|
std::unique_ptr<utils::DispatcherTimer> _security_timer;
|
|
std::unique_ptr<utils::DispatcherTimer> _recovery_timer;
|
|
std::set<UserSymbol> _symbols;
|
|
MBOReader* _mbo;
|
|
MDConsumer _consumer;
|
|
utils::Event<MDConsumer> _event;
|
|
|
|
MDClientImpl(
|
|
utils::Dispatcher* d,
|
|
api::ISyntheticSecurityService* s,
|
|
MBOReader* mbo,
|
|
const std::set<UserSymbol>* symbols):
|
|
_dispatcher(d),
|
|
_ss(s),
|
|
_security_timer(),
|
|
_recovery_timer(),
|
|
_symbols(symbols ? *symbols : get_system_md_symbols()),
|
|
_mbo(mbo),
|
|
_consumer(_cache, _ss, mbo, &_symbols),
|
|
_event()
|
|
{
|
|
if (d) {
|
|
_security_timer =
|
|
std::make_unique<utils::DispatcherTimer>(*d, config::get<RelayOptions>().security_scan_interval(),[this]() { scan_securities(); }),
|
|
_recovery_timer =
|
|
std::make_unique<utils::DispatcherTimer>(*d, config::get<RelayOptions>().recovery_interval(), [this]() { check_heartbeats(); }),
|
|
_security_timer->start();
|
|
_recovery_timer->start();
|
|
}
|
|
}
|
|
|
|
void check_heartbeats() {
|
|
if (!_consumer.all_heartbeats_ok()) {
|
|
MDConsumer temp(_cache,_ss,_mbo,&_symbols);
|
|
if (temp.all_heartbeats_ok()) {
|
|
_consumer.swap(temp);
|
|
LOG(md,Info, "Recovering from bad MD. Old consumer {} new consumer {}", _consumer.id(), temp.id());
|
|
// Subscribers are responsible for keeping the old consumer reference if they need it
|
|
_event.publish(_consumer);
|
|
}
|
|
}
|
|
}
|
|
|
|
size_t scan_securities() {
|
|
if (_consumer.all_heartbeats_ok()) {
|
|
TimeStamp start = TimeStamp::now();
|
|
auto count = _consumer.scan_securities();
|
|
auto delta = (TimeStamp::now() - start);
|
|
if (delta.milliseconds() > 10) {
|
|
LOG(md,Diag,"Scanned handles: {}ms found: {} securities. " ,
|
|
delta.milliseconds(),count);
|
|
}
|
|
return count;
|
|
}
|
|
return 0;
|
|
}
|
|
};
|
|
|
|
|
|
MDClient::MDClient(
|
|
utils::Dispatcher* d,
|
|
api::ISyntheticSecurityService* s,
|
|
MBOReader* mbo,
|
|
const std::set<UserSymbol>* symbols) :
|
|
_self(new MDClientImpl(d,s,mbo,symbols))
|
|
{ }
|
|
|
|
MDClient::~MDClient() { }
|
|
|
|
size_t MDClient::scan_securities() { return _self->scan_securities(); }
|
|
void MDClient::check_heartbeats() { return _self->check_heartbeats(); }
|
|
|
|
utils::EventSubscriber<MDConsumer>& MDClient::on_consumer() {
|
|
return _self->_event;
|
|
}
|
|
MDConsumer& MDClient::current_consumer() const {
|
|
return _self->_consumer;
|
|
}
|
|
SecurityOracle MDClient::oracle() const {
|
|
return SecurityOracle(_self->_cache);
|
|
}
|
|
std::set<UserSymbol> MDClient::symbols() const {
|
|
return _self->_symbols;
|
|
}
|
|
|
|
}}//namespace
|