diff --git a/CMakeLists.txt b/CMakeLists.txt index 9a53d63..a32b870 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -5,6 +5,9 @@ option(LAPS_INSTALL "Install the laps Components" ON) option(LAPS_BUILD_TESTS "Build Tests laps" OFF) +option(LAPS_USE_BROTLI "Enable picotls Brotli TLS certificate compression (libbrotlidec)" OFF) +set(QUICR_USE_BROTLI "${LAPS_USE_BROTLI}" CACHE BOOL "Enable picotls Brotli TLS certificate compression" FORCE) + set(CMAKE_CXX_STANDARD 20) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_CXX_EXTENSIONS OFF) @@ -15,7 +18,7 @@ set (CMAKE_POLICY_VERSION_MINIMUM 3.5) ### Define the project name and version ### Version will be added to version output of apps and library project(laps - VERSION 0.18.16 + VERSION 0.18.23 DESCRIPTION "Latency Aware Publish/Subscribe" LANGUAGES CXX) diff --git a/dependencies/libquicr b/dependencies/libquicr index 0b0841f..6fcf5e5 160000 --- a/dependencies/libquicr +++ b/dependencies/libquicr @@ -1 +1 @@ -Subproject commit 0b0841f4912d88c8227806daaccef115fb4b1aa2 +Subproject commit 6fcf5e5287d2e080a7b517a42faccf3672afc114 diff --git a/src/client_manager.cc b/src/client_manager.cc index 37c158b..04974d9 100644 --- a/src/client_manager.cc +++ b/src/client_manager.cc @@ -271,7 +271,7 @@ namespace laps { } quicr::Reply ClientManager::NewGroupRequested(const quicr::FullTrackName& track_full_name, - std::uint64_t group_id) + std::uint64_t group_id) { auto th = quicr::TrackHash(track_full_name); SPDLOG_INFO("New group requested received track_alais: {} group_id: {} ", th.track_fullname_hash, group_id); @@ -921,8 +921,9 @@ namespace laps { return {}; } - quicr::Reply ClientManager::PublishDoneReceived(const std::shared_ptr& session, - uint64_t request_id) + quicr::Reply ClientManager::PublishDoneReceived( + const std::shared_ptr& session, + uint64_t request_id) { const auto connection_handle = ConnectionHandle(session); SPDLOG_LOGGER_INFO( @@ -1001,6 +1002,25 @@ namespace laps { } } + const auto& pub_ns = s_it->second->GetFullTrackName().name_space; + const auto& pub_ta = s_it->second->GetTrackAlias().value_or(0); + + // UnPublishTrack within publish namespace handler if matched + for (auto& [tn, conns] : state_.subscribes_namespaces) { + const auto prefix_match = tn.IsPrefixOf(pub_ns); + + if (prefix_match == std::partial_ordering::less || prefix_match == std::partial_ordering::equivalent) { + for (auto& [_, pub_ns_h] : conns) { + if (!config_.allow_self && pub_ns_h->GetConnectionId() == connection_handle) { + // Initially do not mirror + continue; + } + + pub_ns_h->UnpublishTrack(pub_ta); + } + } + } + state_.pub_subscribes_by_req_id.erase(s_it); lock.unlock(); @@ -1011,8 +1031,9 @@ namespace laps { return {}; } - quicr::Reply ClientManager::UnsubscribeReceived(const std::shared_ptr& session, - uint64_t request_id) + quicr::Reply ClientManager::UnsubscribeReceived( + const std::shared_ptr& session, + uint64_t request_id) { const auto connection_handle = ConnectionHandle(session); SPDLOG_LOGGER_INFO(LOGGER, "Unsubscribe connection handle: {0} request_id: {1}", connection_handle, request_id); diff --git a/src/client_manager.h b/src/client_manager.h index 5332c1c..bfb22e9 100644 --- a/src/client_manager.h +++ b/src/client_manager.h @@ -28,10 +28,7 @@ namespace laps { quicr::Bytes data; // libc++ now compares set keys with std::less, which requires operator<. - bool operator<(const CacheObject& other) const noexcept - { - return headers.object_id < other.headers.object_id; - } + bool operator<(const CacheObject& other) const noexcept { return headers.object_id < other.headers.object_id; } }; } @@ -196,7 +193,7 @@ namespace laps { std::weak_ptr sub_ns_handler) override; quicr::Reply PublishDoneReceived(const std::shared_ptr& session, - std::uint64_t request_id) override; + std::uint64_t request_id) override; quicr::Reply SubscribeReceived( const std::shared_ptr& session, @@ -205,10 +202,10 @@ namespace laps { const quicr::SubscribeAttributes& subscribe_attributes) override; quicr::Reply UnsubscribeReceived(const std::shared_ptr& session, - std::uint64_t request_id) override; + std::uint64_t request_id) override; quicr::Reply NewGroupRequested(const quicr::FullTrackName& track_full_name, - std::uint64_t group_id) override; + std::uint64_t group_id) override; quicr::Reply TrackStatusReceived( const std::shared_ptr& session, diff --git a/src/peering/info_base.h b/src/peering/info_base.h index 3fa3f98..62e7604 100644 --- a/src/peering/info_base.h +++ b/src/peering/info_base.h @@ -8,7 +8,7 @@ #include "peering/messages/subscribe_info.h" #include -#include +#include #include namespace laps::peering { diff --git a/src/peering/messages/data_header.h b/src/peering/messages/data_header.h index 5e2bb02..1327749 100644 --- a/src/peering/messages/data_header.h +++ b/src/peering/messages/data_header.h @@ -5,7 +5,7 @@ #include "peering/common.h" #include "subscribe_info.h" -#include +#include #include namespace laps::peering { diff --git a/src/peering/peer_manager.cc b/src/peering/peer_manager.cc index 0451b0b..df3e7e7 100644 --- a/src/peering/peer_manager.cc +++ b/src/peering/peer_manager.cc @@ -1364,8 +1364,7 @@ namespace laps::peering { if (const auto stream_it = entry.streams.find(stream_id); stream_it != entry.streams.end()) { out_peer_sess->CloseStream(stream_it->second, flag); entry.streams.erase(stream_it); - client_manager_->PeerStreamClosed( - key.first, stream_id, flag == quicr::StreamClosedFlag::kReset); + client_manager_->PeerStreamClosed(key.first, stream_id, flag == quicr::StreamClosedFlag::kReset); } } } diff --git a/src/publish_namespace_handler.cc b/src/publish_namespace_handler.cc index 3974733..6e32f75 100644 --- a/src/publish_namespace_handler.cc +++ b/src/publish_namespace_handler.cc @@ -15,6 +15,12 @@ namespace laps { } } +void +laps::PublishNamespaceHandler::UnpublishTrack(std::uint64_t track_alias) +{ + published_tracks_.erase(track_alias); +} + void laps::PublishNamespaceHandler::PublishTrack(std::shared_ptr handler) { @@ -152,8 +158,6 @@ laps::PublishNamespaceHandler::UpdateTrackRanking( if (published_tracks_.size() > active_tracks.size()) { - std::vector remove_publish_tracks; - // Unpublish tracks that are too old for (auto& [ta, track] : published_tracks_) { if (active_tracks.contains(ta)) { @@ -182,21 +186,14 @@ laps::PublishNamespaceHandler::UpdateTrackRanking( } if (cur_tick - track.last_updated_tick > delay_publish_done_ms_) { - SPDLOG_INFO("Unpublish track, not in top-n track alias: {} conn_id: {} ticks: {} < {}", - ta, - GetConnectionId(), - track.last_updated_tick, - cur_tick); - remove_publish_tracks.push_back(ta); + SPDLOG_TRACE("Unpublish track, not in top-n track alias: {} conn_id: {} ticks: {} < {}", + ta, + GetConnectionId(), + track.last_updated_tick, + cur_tick); } } } - - if (!remove_publish_tracks.empty()) { - for (const auto& ta: remove_publish_tracks) { - published_tracks_.erase(ta); - } - } } } diff --git a/src/publish_namespace_handler.h b/src/publish_namespace_handler.h index fb37f05..0e154ef 100644 --- a/src/publish_namespace_handler.h +++ b/src/publish_namespace_handler.h @@ -44,6 +44,7 @@ namespace laps { std::shared_ptr> data) override; void PublishTrack(std::shared_ptr handler) override; + void UnpublishTrack(std::uint64_t track_alias); static auto Create(const quicr::TrackNamespace& prefix, std::weak_ptr tick_service) { diff --git a/src/subscribe_handler.cc b/src/subscribe_handler.cc index d466858..6530707 100644 --- a/src/subscribe_handler.cc +++ b/src/subscribe_handler.cc @@ -140,6 +140,9 @@ namespace laps { } if (value.latest_value != recv_value || cur_tick - value.latest_tick_ms > kRefreshRankingIntervalMs) { + SPDLOG_TRACE( + "value changed: from {} to {} ta: {} conn: {}", value.latest_value, recv_value, ta, conn_id); + value.latest_value = recv_value; value.latest_tick_ms = cur_tick; diff --git a/src/subscribe_handler.h b/src/subscribe_handler.h index 3974666..753a383 100644 --- a/src/subscribe_handler.h +++ b/src/subscribe_handler.h @@ -16,7 +16,7 @@ namespace laps { class SubscribeTrackHandler : public quicr::SubscribeTrackHandler { public: - static constexpr uint64_t kRefreshRankingIntervalMs = 120; + static constexpr uint64_t kRefreshRankingIntervalMs = 500; SubscribeTrackHandler(const quicr::FullTrackName& full_track_name, std::uint8_t priority, diff --git a/src/track_ranking.h b/src/track_ranking.h index c8ce582..aa7c597 100644 --- a/src/track_ranking.h +++ b/src/track_ranking.h @@ -132,6 +132,8 @@ namespace laps { flat_track_list_.insert(flat_track_list_.end(), sort_tracks.begin(), sort_tracks.end()); } + NotifyNamespaceHandlers(); + } else { // Update latest_tick in flat_track_list_ in-place for (auto& [alias, insert_seq_num, latest_tick, conn_id] : flat_track_list_) { @@ -142,8 +144,6 @@ namespace laps { } } } - - NotifyNamespaceHandlers(); } void RemoveTrack(const TrackAlias track_alias)