From 2472dc2268c4c62ab5452a2c6e1b84d6c102d3be Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Thu, 3 Sep 2026 04:35:44 +0300 Subject: [PATCH] refactor(market-data): split router public headers --- .../market_data/MarketDataRouter.hpp | 195 +------------ .../market_data/MarketDataRouterIds.hpp | 121 ++++++++ .../MarketDataRouterSubscription.hpp | 102 +++++++ .../market_data/detail/MarketDataRouter.ipp | 273 ++++++++---------- 4 files changed, 352 insertions(+), 339 deletions(-) create mode 100644 include/optionx_cpp/market_data/MarketDataRouterIds.hpp create mode 100644 include/optionx_cpp/market_data/MarketDataRouterSubscription.hpp diff --git a/include/optionx_cpp/market_data/MarketDataRouter.hpp b/include/optionx_cpp/market_data/MarketDataRouter.hpp index ee61ca3..57b5fac 100644 --- a/include/optionx_cpp/market_data/MarketDataRouter.hpp +++ b/include/optionx_cpp/market_data/MarketDataRouter.hpp @@ -3,7 +3,10 @@ #define OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_HPP_INCLUDED /// \file MarketDataRouter.hpp -/// \brief Defines subscription-scoped market-data routing utilities. +/// \brief Defines the subscription-scoped market-data Router facade. + +#include "MarketDataRouterIds.hpp" +#include "MarketDataRouterSubscription.hpp" namespace optionx::market_data { @@ -11,196 +14,6 @@ namespace optionx::market_data { class MarketDataRouterState; } // namespace detail - /// \class RoutedSubscriptionId - /// \brief Strong Router-local identifier of a logical market-data subscription. - class RoutedSubscriptionId { - public: - /// \brief Constructs an invalid identifier. - constexpr RoutedSubscriptionId() noexcept = default; - - /// \brief Returns true when the Router assigned an identifier. - [[nodiscard]] constexpr bool valid() const noexcept { - return m_value != 0; - } - - /// \brief Allows identifiers to be checked directly in conditions. - constexpr explicit operator bool() const noexcept { - return valid(); - } - - /// \brief Returns the numeric value for logging and diagnostics. - [[nodiscard]] constexpr std::uint64_t value() const noexcept { - return m_value; - } - - friend constexpr bool operator==( - RoutedSubscriptionId lhs, - RoutedSubscriptionId rhs) noexcept { - return lhs.m_value == rhs.m_value; - } - - friend constexpr bool operator!=( - RoutedSubscriptionId lhs, - RoutedSubscriptionId rhs) noexcept { - return !(lhs == rhs); - } - - private: - std::uint64_t m_value = 0; - - explicit constexpr RoutedSubscriptionId(std::uint64_t value) noexcept - : m_value(value) {} - - friend class detail::MarketDataRouterState; - }; - - /// \struct RoutedSubscriptionIdHash - /// \brief Hashes a routed subscription identifier for unordered containers. - struct RoutedSubscriptionIdHash { - [[nodiscard]] std::size_t operator()(RoutedSubscriptionId id) const noexcept { - return std::hash{}(id.value()); - } - }; - - /// \class MarketDataProviderId - /// \brief Stable application-assigned identifier of a registered provider. - class MarketDataProviderId { - public: - /// \brief Constructs an invalid provider identifier. - constexpr MarketDataProviderId() noexcept = default; - - /// \brief Constructs a stable provider identifier from application data. - explicit constexpr MarketDataProviderId(std::uint64_t value) noexcept - : m_value(value) {} - - /// \brief Returns true when the identifier can select a provider. - [[nodiscard]] constexpr bool valid() const noexcept { - return m_value != 0; - } - - /// \brief Allows identifiers to be checked directly in conditions. - constexpr explicit operator bool() const noexcept { - return valid(); - } - - /// \brief Returns the application-assigned numeric value. - [[nodiscard]] constexpr std::uint64_t value() const noexcept { - return m_value; - } - - friend constexpr bool operator==( - MarketDataProviderId lhs, - MarketDataProviderId rhs) noexcept { - return lhs.m_value == rhs.m_value; - } - - friend constexpr bool operator!=( - MarketDataProviderId lhs, - MarketDataProviderId rhs) noexcept { - return !(lhs == rhs); - } - - private: - std::uint64_t m_value = 0; - }; - - /// \struct MarketDataProviderIdHash - /// \brief Hashes an application-assigned provider identifier. - struct MarketDataProviderIdHash { - [[nodiscard]] std::size_t operator()(MarketDataProviderId id) const noexcept { - return std::hash{}(id.value()); - } - }; - - namespace detail { - - struct MarketDataRouterSubscriptionControl { - mutable std::mutex mutex; - RoutedSubscriptionId router_id; - MarketDataProviderId registered_provider_id; - MarketDataSubscriptionHandle provider_subscription; - std::weak_ptr router; - bool active = false; - bool released = false; - }; - } // namespace detail - - /// \class MarketDataRouterSubscription - /// \brief Move-only RAII owner of one subscription created by MarketDataRouter. - /// \details Destroying or resetting the handle requests unsubscription. The - /// handle is created before an asynchronous provider necessarily - /// accepts the subscription, so valid() can be true while pending() - /// is also true. provider_subscription() becomes valid after the - /// provider reports SUBSCRIBED. - class MarketDataRouterSubscription { - public: - /// \brief Default-constructs an empty handle. - MarketDataRouterSubscription() = default; - - /// \brief Copy construction is disabled because ownership is unique. - MarketDataRouterSubscription(const MarketDataRouterSubscription&) = delete; - /// \brief Copy assignment is disabled because ownership is unique. - MarketDataRouterSubscription& operator=(const MarketDataRouterSubscription&) = delete; - - /// \brief Transfers subscription ownership. - MarketDataRouterSubscription(MarketDataRouterSubscription&& other) noexcept - : m_control(std::move(other.m_control)) {} - - /// \brief Releases the current subscription and transfers ownership. - MarketDataRouterSubscription& operator=(MarketDataRouterSubscription&& other) noexcept; - - /// \brief Requests unsubscription for an owned subscription. - ~MarketDataRouterSubscription() { - reset(); - } - - /// \brief Returns the stable router-local subscription ID. - [[nodiscard]] RoutedSubscriptionId router_id() const noexcept; - - /// \brief Returns the provider-assigned subscription descriptor, if accepted. - [[nodiscard]] MarketDataSubscriptionHandle provider_subscription() const; - - /// \brief Returns the stable registered provider ID used to create this route. - /// \details Direct provider-reference subscriptions return an invalid ID. - [[nodiscard]] MarketDataProviderId registered_provider_id() const; - - /// \brief Returns true while this object owns a pending or active route. - [[nodiscard]] bool valid() const; - - /// \brief Returns true after the provider accepted the subscription. - [[nodiscard]] bool active() const; - - /// \brief Returns true while provider acceptance is still pending. - [[nodiscard]] bool pending() const { - return valid() && !active(); - } - - /// \brief Allows handles to be used in boolean contexts. - explicit operator bool() const { - return valid(); - } - - /// \brief Explicitly releases the route and requests provider unsubscription. - /// \param callback Optional callback receiving the provider unsubscribe result. - /// \return True when an unsubscribe or pending cancellation was accepted. - /// \details The logical route is released even when physical provider cleanup - /// fails. MarketDataRouter retains failed cleanup ownership for retry. - bool unsubscribe(BaseMarketDataProvider::subscription_callback_t callback = {}); - - /// \brief Releases the route without an unsubscribe completion callback. - void reset() noexcept; - - private: - std::shared_ptr m_control; - - explicit MarketDataRouterSubscription( - std::shared_ptr control) - : m_control(std::move(control)) {} - - friend class MarketDataRouter; - friend class detail::MarketDataRouterState; - }; - /// \class MarketDataRouter /// \brief Binds provider subscriptions to concrete subscriber objects. /// \details Unlike MarketDataHub, the router owns provider subscription diff --git a/include/optionx_cpp/market_data/MarketDataRouterIds.hpp b/include/optionx_cpp/market_data/MarketDataRouterIds.hpp new file mode 100644 index 0000000..4df66fe --- /dev/null +++ b/include/optionx_cpp/market_data/MarketDataRouterIds.hpp @@ -0,0 +1,121 @@ +#pragma once +#ifndef OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_IDS_HPP_INCLUDED +#define OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_IDS_HPP_INCLUDED + +/// \file MarketDataRouterIds.hpp +/// \brief Defines the strong identifiers used by MarketDataRouter. + +#include +#include +#include + +namespace optionx::market_data { + + namespace detail { + class MarketDataRouterState; + } // namespace detail + + /// \class RoutedSubscriptionId + /// \brief Strong Router-local identifier of a logical market-data subscription. + class RoutedSubscriptionId { + public: + /// \brief Constructs an invalid identifier. + constexpr RoutedSubscriptionId() noexcept = default; + + /// \brief Returns true when the Router assigned an identifier. + [[nodiscard]] constexpr bool valid() const noexcept { + return m_value != 0; + } + + /// \brief Allows identifiers to be checked directly in conditions. + constexpr explicit operator bool() const noexcept { + return valid(); + } + + /// \brief Returns the numeric value for logging and diagnostics. + [[nodiscard]] constexpr std::uint64_t value() const noexcept { + return m_value; + } + + friend constexpr bool operator==( + RoutedSubscriptionId lhs, + RoutedSubscriptionId rhs) noexcept { + return lhs.m_value == rhs.m_value; + } + + friend constexpr bool operator!=( + RoutedSubscriptionId lhs, + RoutedSubscriptionId rhs) noexcept { + return !(lhs == rhs); + } + + private: + std::uint64_t m_value = 0; + + explicit constexpr RoutedSubscriptionId(std::uint64_t value) noexcept + : m_value(value) {} + + friend class detail::MarketDataRouterState; + }; + + /// \struct RoutedSubscriptionIdHash + /// \brief Hashes a routed subscription identifier for unordered containers. + struct RoutedSubscriptionIdHash { + [[nodiscard]] std::size_t operator()(RoutedSubscriptionId id) const noexcept { + return std::hash{}(id.value()); + } + }; + + /// \class MarketDataProviderId + /// \brief Stable application-assigned identifier of a registered provider. + class MarketDataProviderId { + public: + /// \brief Constructs an invalid provider identifier. + constexpr MarketDataProviderId() noexcept = default; + + /// \brief Constructs a stable provider identifier from application data. + explicit constexpr MarketDataProviderId(std::uint64_t value) noexcept + : m_value(value) {} + + /// \brief Returns true when the identifier can select a provider. + [[nodiscard]] constexpr bool valid() const noexcept { + return m_value != 0; + } + + /// \brief Allows identifiers to be checked directly in conditions. + constexpr explicit operator bool() const noexcept { + return valid(); + } + + /// \brief Returns the application-assigned numeric value. + [[nodiscard]] constexpr std::uint64_t value() const noexcept { + return m_value; + } + + friend constexpr bool operator==( + MarketDataProviderId lhs, + MarketDataProviderId rhs) noexcept { + return lhs.m_value == rhs.m_value; + } + + friend constexpr bool operator!=( + MarketDataProviderId lhs, + MarketDataProviderId rhs) noexcept { + return !(lhs == rhs); + } + + private: + std::uint64_t m_value = 0; + }; + + /// \struct MarketDataProviderIdHash + /// \brief Hashes an application-assigned provider identifier. + struct MarketDataProviderIdHash { + [[nodiscard]] std::size_t operator()(MarketDataProviderId id) const noexcept { + return std::hash{}(id.value()); + } + }; + +} // namespace optionx::market_data + +#endif // OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_IDS_HPP_INCLUDED diff --git a/include/optionx_cpp/market_data/MarketDataRouterSubscription.hpp b/include/optionx_cpp/market_data/MarketDataRouterSubscription.hpp new file mode 100644 index 0000000..4feb3ac --- /dev/null +++ b/include/optionx_cpp/market_data/MarketDataRouterSubscription.hpp @@ -0,0 +1,102 @@ +#pragma once +#ifndef OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_SUBSCRIPTION_HPP_INCLUDED +#define OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_SUBSCRIPTION_HPP_INCLUDED + +/// \file MarketDataRouterSubscription.hpp +/// \brief Defines the move-only RAII handle returned by MarketDataRouter. + +#include +#include +#include + +#include "BaseMarketDataProvider.hpp" +#include "MarketDataRouterIds.hpp" + +namespace optionx::market_data { + + class MarketDataRouter; + + namespace detail { + class MarketDataRouterState; + struct MarketDataRouterSubscriptionControl; + } // namespace detail + + /// \class MarketDataRouterSubscription + /// \brief Move-only RAII owner of one subscription created by MarketDataRouter. + /// \details Destroying or resetting the handle requests unsubscription. The + /// handle is created before an asynchronous provider necessarily + /// accepts the subscription, so valid() can be true while pending() + /// is also true. provider_subscription() becomes valid after the + /// provider reports SUBSCRIBED. + class MarketDataRouterSubscription { + public: + /// \brief Default-constructs an empty handle. + MarketDataRouterSubscription() = default; + + /// \brief Copy construction is disabled because ownership is unique. + MarketDataRouterSubscription(const MarketDataRouterSubscription&) = delete; + /// \brief Copy assignment is disabled because ownership is unique. + MarketDataRouterSubscription& operator=(const MarketDataRouterSubscription&) = delete; + + /// \brief Transfers subscription ownership. + MarketDataRouterSubscription(MarketDataRouterSubscription&& other) noexcept + : m_control(std::move(other.m_control)) {} + + /// \brief Releases the current subscription and transfers ownership. + MarketDataRouterSubscription& operator=(MarketDataRouterSubscription&& other) noexcept; + + /// \brief Requests unsubscription for an owned subscription. + ~MarketDataRouterSubscription() { + reset(); + } + + /// \brief Returns the stable router-local subscription ID. + [[nodiscard]] RoutedSubscriptionId router_id() const noexcept; + + /// \brief Returns the provider-assigned subscription descriptor, if accepted. + [[nodiscard]] MarketDataSubscriptionHandle provider_subscription() const; + + /// \brief Returns the stable registered provider ID used to create this route. + /// \details Direct provider-reference subscriptions return an invalid ID. + [[nodiscard]] MarketDataProviderId registered_provider_id() const; + + /// \brief Returns true while this object owns a pending or active route. + [[nodiscard]] bool valid() const; + + /// \brief Returns true after the provider accepted the subscription. + [[nodiscard]] bool active() const; + + /// \brief Returns true while provider acceptance is still pending. + [[nodiscard]] bool pending() const { + return valid() && !active(); + } + + /// \brief Allows handles to be used in boolean contexts. + explicit operator bool() const { + return valid(); + } + + /// \brief Explicitly releases the route and requests provider unsubscription. + /// \param callback Optional callback receiving the provider unsubscribe result. + /// \return True when an unsubscribe or pending cancellation was accepted. + /// \details The logical route is released even when physical provider cleanup + /// fails. MarketDataRouter retains failed cleanup ownership for retry. + bool unsubscribe(BaseMarketDataProvider::subscription_callback_t callback = {}); + + /// \brief Releases the route without an unsubscribe completion callback. + void reset() noexcept; + + private: + std::shared_ptr m_control; + + explicit MarketDataRouterSubscription( + std::shared_ptr control) + : m_control(std::move(control)) {} + + friend class MarketDataRouter; + friend class detail::MarketDataRouterState; + }; + +} // namespace optionx::market_data + +#endif // OPTIONX_HEADER_MARKET_DATA_MARKET_DATA_ROUTER_SUBSCRIPTION_HPP_INCLUDED diff --git a/include/optionx_cpp/market_data/detail/MarketDataRouter.ipp b/include/optionx_cpp/market_data/detail/MarketDataRouter.ipp index b9b8ebf..592280c 100644 --- a/include/optionx_cpp/market_data/detail/MarketDataRouter.ipp +++ b/include/optionx_cpp/market_data/detail/MarketDataRouter.ipp @@ -9,6 +9,16 @@ namespace optionx::market_data { namespace detail { + struct MarketDataRouterSubscriptionControl { + mutable std::mutex mutex; + RoutedSubscriptionId router_id; + MarketDataProviderId registered_provider_id; + MarketDataSubscriptionHandle provider_subscription; + std::weak_ptr router; + bool active = false; + bool released = false; + }; + class MarketDataRouterState final : public std::enable_shared_from_this { public: @@ -119,6 +129,18 @@ namespace optionx::market_data { BarSubscriptionRequest request, subscription_callback_t callback); + template + MarketDataRouterSubscription subscribe_impl( + BaseMarketDataProvider& provider, + std::weak_ptr subscriber, + Request request, + subscription_callback_t callback, + MarketDataProviderId registered_provider_id, + SubscribeOperation subscribe_operation, + const char* invalid_request_message, + const char* operation_name, + const char* not_accepted_message); + bool unsubscribe( const std::shared_ptr& control, subscription_callback_t callback); @@ -145,6 +167,13 @@ namespace optionx::market_data { ProviderInstanceId provider_id, MarketDataStatusUpdate update); + template + void route_batch( + ProviderInstanceId provider_id, + std::unique_ptr batch, + MatchesStream matches_stream, + Deliver deliver); + private: mutable std::mutex m_mutex; std::unordered_map< @@ -720,19 +749,25 @@ namespace optionx::market_data { return control; } - inline MarketDataRouterSubscription MarketDataRouterState::subscribe_ticks( + template + inline MarketDataRouterSubscription + MarketDataRouterState::subscribe_impl( BaseMarketDataProvider& provider, std::weak_ptr subscriber, - TickSubscriptionRequest request, + Request request, subscription_callback_t callback, - MarketDataProviderId registered_provider_id) { + MarketDataProviderId registered_provider_id, + SubscribeOperation subscribe_operation, + const char* invalid_request_message, + const char* operation_name, + const char* not_accepted_message) { if (!request.valid()) { dispatch_result( std::move(callback), MarketDataSubscriptionResult::failed( std::move(request), MarketDataSubscriptionStatus::INVALID_REQUEST, - "Invalid tick subscription request.")); + invalid_request_message)); return {}; } @@ -758,7 +793,8 @@ namespace optionx::market_data { const auto state = shared_from_this(); bool accepted = false; try { - accepted = provider.subscribe_ticks( + accepted = subscribe_operation( + provider, std::move(request), [state, router_id, &provider, callback]( MarketDataSubscriptionResult result) mutable { @@ -775,8 +811,7 @@ namespace optionx::market_data { MarketDataSubscriptionResult::failed( request_for_failure, MarketDataSubscriptionStatus::FAILED, - std::string("Market-data provider tick subscription threw: ") + - exception.what()), + std::string(operation_name) + " threw: " + exception.what()), std::move(callback)); return MarketDataRouterSubscription(std::move(control)); } catch (...) { @@ -786,7 +821,7 @@ namespace optionx::market_data { MarketDataSubscriptionResult::failed( request_for_failure, MarketDataSubscriptionStatus::FAILED, - "Market-data provider tick subscription threw."), + std::string(operation_name) + " threw."), std::move(callback)); return MarketDataRouterSubscription(std::move(control)); } @@ -798,12 +833,36 @@ namespace optionx::market_data { MarketDataSubscriptionResult::failed( request_for_failure, MarketDataSubscriptionStatus::FAILED, - "Market-data provider did not accept the tick subscription operation."), + not_accepted_message), std::move(callback)); } return MarketDataRouterSubscription(std::move(control)); } + inline MarketDataRouterSubscription MarketDataRouterState::subscribe_ticks( + BaseMarketDataProvider& provider, + std::weak_ptr subscriber, + TickSubscriptionRequest request, + subscription_callback_t callback, + MarketDataProviderId registered_provider_id) { + return subscribe_impl( + provider, + std::move(subscriber), + std::move(request), + std::move(callback), + registered_provider_id, + [](BaseMarketDataProvider& provider, + TickSubscriptionRequest request, + subscription_callback_t operation_callback) { + return provider.subscribe_ticks( + std::move(request), + std::move(operation_callback)); + }, + "Invalid tick subscription request.", + "Market-data provider tick subscription", + "Market-data provider did not accept the tick subscription operation."); + } + inline MarketDataRouterSubscription MarketDataRouterState::subscribe_ticks( MarketDataProviderId provider_id, std::weak_ptr subscriber, @@ -865,82 +924,22 @@ namespace optionx::market_data { BarSubscriptionRequest request, subscription_callback_t callback, MarketDataProviderId registered_provider_id) { - if (!request.valid()) { - dispatch_result( - std::move(callback), - MarketDataSubscriptionResult::failed( - std::move(request), - MarketDataSubscriptionStatus::INVALID_REQUEST, - "Invalid bar subscription request.")); - return {}; - } - - const auto request_for_failure = request; - std::string error_message; - auto control = add_pending_entry( + return subscribe_impl( provider, std::move(subscriber), - stream_from(request), + std::move(request), + std::move(callback), registered_provider_id, - error_message); - if (!control) { - dispatch_result( - std::move(callback), - MarketDataSubscriptionResult::failed( - request_for_failure, - MarketDataSubscriptionStatus::FAILED, - std::move(error_message))); - return {}; - } - - const auto router_id = control->router_id; - const auto state = shared_from_this(); - bool accepted = false; - try { - accepted = provider.subscribe_bars( - std::move(request), - [state, router_id, &provider, callback]( - MarketDataSubscriptionResult result) mutable { - state->dispatch_subscribe_completion( - router_id, - provider, - std::move(result), - std::move(callback)); - }); - } catch (const std::exception& exception) { - fail_pending_subscribe( - router_id, - provider, - MarketDataSubscriptionResult::failed( - request_for_failure, - MarketDataSubscriptionStatus::FAILED, - std::string("Market-data provider bar subscription threw: ") + - exception.what()), - std::move(callback)); - return MarketDataRouterSubscription(std::move(control)); - } catch (...) { - fail_pending_subscribe( - router_id, - provider, - MarketDataSubscriptionResult::failed( - request_for_failure, - MarketDataSubscriptionStatus::FAILED, - "Market-data provider bar subscription threw."), - std::move(callback)); - return MarketDataRouterSubscription(std::move(control)); - } - - if (!accepted) { - fail_pending_subscribe( - router_id, - provider, - MarketDataSubscriptionResult::failed( - request_for_failure, - MarketDataSubscriptionStatus::FAILED, - "Market-data provider did not accept the bar subscription operation."), - std::move(callback)); - } - return MarketDataRouterSubscription(std::move(control)); + [](BaseMarketDataProvider& provider, + BarSubscriptionRequest request, + subscription_callback_t operation_callback) { + return provider.subscribe_bars( + std::move(request), + std::move(operation_callback)); + }, + "Invalid bar subscription request.", + "Market-data provider bar subscription", + "Market-data provider did not accept the bar subscription operation."); } inline MarketDataRouterSubscription MarketDataRouterState::subscribe_bars( @@ -1374,64 +1373,30 @@ namespace optionx::market_data { return true; } - inline void MarketDataRouterState::route_ticks( + template + inline void MarketDataRouterState::route_batch( ProviderInstanceId provider_id, - std::unique_ptr batch) { + std::unique_ptr batch, + MatchesStream matches_stream, + Deliver deliver) { if (!batch) return; - std::vector, TickDataBatch>> deliveries; + std::vector, Batch>> deliveries; { std::lock_guard lock(m_mutex); const auto provider_it = m_providers.find(provider_id); if (provider_it == m_providers.end()) return; - if (batch->subscription.valid()) { - if (batch->subscription.provider_id != provider_id) return; - const auto route_it = provider_it->second.provider_routes.find( - batch->subscription.id); - if (route_it == provider_it->second.provider_routes.end()) return; - const auto entry_it = m_entries.find(route_it->second); - if (entry_it == m_entries.end()) return; - const auto& entry = entry_it->second; - auto subscriber = entry->subscriber.lock(); - if (!subscriber || - entry->phase != EntryPhase::ACTIVE || - !batch_matches_stream(*batch, entry->stream)) { + auto add_delivery = [&](const std::shared_ptr& entry) { + if (entry->phase != EntryPhase::ACTIVE || + !matches_stream(*batch, entry->stream)) { return; } + auto subscriber = entry->subscriber.lock(); + if (!subscriber) return; auto routed = *batch; routed.subscription = entry->control->provider_subscription; deliveries.emplace_back(std::move(subscriber), std::move(routed)); - } else { - for (const auto& [id, entry] : m_entries) { - (void)id; - if (entry->provider_id != provider_id || - entry->phase != EntryPhase::ACTIVE || - !batch_matches_stream(*batch, entry->stream)) { - continue; - } - auto subscriber = entry->subscriber.lock(); - if (!subscriber) continue; - auto routed = *batch; - routed.subscription = entry->control->provider_subscription; - deliveries.emplace_back(std::move(subscriber), std::move(routed)); - } - } - } - - for (auto& delivery : deliveries) { - delivery.first->on_tick_data(delivery.second); - } - } - - inline void MarketDataRouterState::route_bars( - ProviderInstanceId provider_id, - std::unique_ptr batch) { - if (!batch) return; - std::vector, BarDataBatch>> deliveries; - { - std::lock_guard lock(m_mutex); - const auto provider_it = m_providers.find(provider_id); - if (provider_it == m_providers.end()) return; + }; if (batch->subscription.valid()) { if (batch->subscription.provider_id != provider_id) return; @@ -1440,38 +1405,50 @@ namespace optionx::market_data { if (route_it == provider_it->second.provider_routes.end()) return; const auto entry_it = m_entries.find(route_it->second); if (entry_it == m_entries.end()) return; - const auto& entry = entry_it->second; - auto subscriber = entry->subscriber.lock(); - if (!subscriber || - entry->phase != EntryPhase::ACTIVE || - !batch_matches_stream(*batch, entry->stream)) { - return; - } - auto routed = *batch; - routed.subscription = entry->control->provider_subscription; - deliveries.emplace_back(std::move(subscriber), std::move(routed)); + add_delivery(entry_it->second); } else { for (const auto& [id, entry] : m_entries) { (void)id; - if (entry->provider_id != provider_id || - entry->phase != EntryPhase::ACTIVE || - !batch_matches_stream(*batch, entry->stream)) { - continue; + if (entry->provider_id == provider_id) { + add_delivery(entry); } - auto subscriber = entry->subscriber.lock(); - if (!subscriber) continue; - auto routed = *batch; - routed.subscription = entry->control->provider_subscription; - deliveries.emplace_back(std::move(subscriber), std::move(routed)); } } } for (auto& delivery : deliveries) { - delivery.first->on_bar_data(delivery.second); + deliver(*delivery.first, delivery.second); } } + inline void MarketDataRouterState::route_ticks( + ProviderInstanceId provider_id, + std::unique_ptr batch) { + route_batch( + provider_id, + std::move(batch), + [this](const TickDataBatch& data, const StreamDescriptor& stream) { + return batch_matches_stream(data, stream); + }, + [](IMarketDataSubscriber& subscriber, TickDataBatch& data) { + subscriber.on_tick_data(data); + }); + } + + inline void MarketDataRouterState::route_bars( + ProviderInstanceId provider_id, + std::unique_ptr batch) { + route_batch( + provider_id, + std::move(batch), + [this](const BarDataBatch& data, const StreamDescriptor& stream) { + return batch_matches_stream(data, stream); + }, + [](IMarketDataSubscriber& subscriber, BarDataBatch& data) { + subscriber.on_bar_data(data); + }); + } + inline void MarketDataRouterState::route_status( ProviderInstanceId provider_id, MarketDataStatusUpdate update) {