diff --git a/src/MoqxRelay.cpp b/src/MoqxRelay.cpp index 9eaa63fb..c467b44e 100644 --- a/src/MoqxRelay.cpp +++ b/src/MoqxRelay.cpp @@ -454,30 +454,24 @@ MoqxRelay::validatePublishNamespace(const FullTrackName& ftn, RequestID requestI return std::nullopt; } -// Constructs the publisher's local forwarder and installs its callback chain (Weak -> -// CrossExec -> LocalForwarder) on publisherExec, before the reply hops to relayExec_. -// tlForwarders_ must already be initialized. -std::shared_ptr MoqxRelay::createPublisherForwarder(const PublishRequest& pub) { - const auto& ftn = pub.fullTrackName; - // Returned uninitialized; the caller's Claim applies the initial state. - auto localPubFwd = std::make_shared(ftn); - - // removeOnEmpty=false: the publisher's forwarder must survive subscriber churn, so - // LocalForwarderCallback removes it from tlForwarders_ only when the source ends. +// Chain and entry claim are one call: LocalForwarderCallback vacates the entry when the +// source ends, so a chain over a forwarder that does not hold it has nothing to remove. +LocalForwarderRegistry::Claim MoqxRelay::installPublisherForwarder( + const FullTrackName& ftn, + const std::shared_ptr& fwd +) { + // removeOnEmpty=false: the publisher's forwarder must survive subscriber churn. auto relayAdapter = std::make_shared(weak_from_this()); - auto crossExec = std::make_shared( - relayExec_, - localPubFwd, - std::move(relayAdapter) - ); - localPubFwd->setCallback(std::make_shared( + auto crossExec = + std::make_shared(relayExec_, fwd, std::move(relayAdapter)); + fwd->setCallback(std::make_shared( tlForwarders_.get(), ftn, std::move(crossExec), /*removeOnEmpty=*/false )); - return localPubFwd; + return tlForwarders_->replace(ftn, fwd); } // Called from LocalPublishFilter::publish() on publisherExec. Creates the publisher's @@ -495,12 +489,8 @@ Subscriber::PublishResult MoqxRelay::publishFromPublisherExec( tlForwarders_.reset(new LocalForwarderRegistry()); } - auto localPubFwd = createPublisherForwarder(pub); - - // The publisher's forwarder is authoritative — claim the entry, displacing any - // stale subscribe-path local forwarder so same-thread subscribers reuse THIS - // forwarder via the fast path. - tlForwarders_->replace(pub.fullTrackName, localPubFwd) + auto localPubFwd = std::make_shared(pub.fullTrackName); + installPublisherForwarder(pub.fullTrackName, localPubFwd) .markReady(InitialTrackState{pub.largest, pub.extensions}); // crossExecFilter is a channel subscriber for the relay exec diff --git a/src/MoqxRelay.h b/src/MoqxRelay.h index 5762212e..8d71430e 100644 --- a/src/MoqxRelay.h +++ b/src/MoqxRelay.h @@ -363,9 +363,10 @@ class MoqxRelay : public moxygen::Publisher, std::shared_ptr session ); - // Constructs the publisher's local forwarder and installs its callback chain on - // publisherExec. tlForwarders_ must already be initialized. - std::shared_ptr createPublisherForwarder(const moxygen::PublishRequest& pub + // Runs on fwd's exec; the returned Claim owes a markReady/fail. + LocalForwarderRegistry::Claim installPublisherForwarder( + const moxygen::FullTrackName& ftn, + const std::shared_ptr& fwd ); std::optional