Skip to content

Commit d194100

Browse files
afrindclaude
andcommitted
relay: fold registry replacement into installPublisherForwarder
Previously, createPublisherForwarder built a forwarder and returned it to the caller to register. Now, installPublisherForwarder takes an already-constructed forwarder, wires its callback chain, and returns the LocalForwarderRegistry::Claim from replacing the registry entry, so the subscribe path can use it in a subsequent commit. publishFromPublisherExec constructs the forwarder itself and calls markReady on that claim. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent 1f8d73c commit d194100

2 files changed

Lines changed: 17 additions & 26 deletions

File tree

src/MoqxRelay.cpp

Lines changed: 13 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -454,30 +454,24 @@ MoqxRelay::validatePublishNamespace(const FullTrackName& ftn, RequestID requestI
454454
return std::nullopt;
455455
}
456456

457-
// Constructs the publisher's local forwarder and installs its callback chain (Weak ->
458-
// CrossExec -> LocalForwarder) on publisherExec, before the reply hops to relayExec_.
459-
// tlForwarders_ must already be initialized.
460-
std::shared_ptr<MoQForwarder> MoqxRelay::createPublisherForwarder(const PublishRequest& pub) {
461-
const auto& ftn = pub.fullTrackName;
462-
// Returned uninitialized; the caller's Claim applies the initial state.
463-
auto localPubFwd = std::make_shared<MoQForwarder>(ftn);
464-
465-
// removeOnEmpty=false: the publisher's forwarder must survive subscriber churn, so
466-
// LocalForwarderCallback removes it from tlForwarders_ only when the source ends.
457+
// Chain and entry claim are one call: LocalForwarderCallback vacates the entry when the
458+
// source ends, so a chain over a forwarder that does not hold it has nothing to remove.
459+
LocalForwarderRegistry::Claim MoqxRelay::installPublisherForwarder(
460+
const FullTrackName& ftn,
461+
const std::shared_ptr<MoQForwarder>& fwd
462+
) {
463+
// removeOnEmpty=false: the publisher's forwarder must survive subscriber churn.
467464
auto relayAdapter = std::make_shared<WeakRelayForwarderCallback>(weak_from_this());
468-
auto crossExec = std::make_shared<CrossExecForwarderCallback>(
469-
relayExec_,
470-
localPubFwd,
471-
std::move(relayAdapter)
472-
);
473-
localPubFwd->setCallback(std::make_shared<LocalForwarderCallback>(
465+
auto crossExec =
466+
std::make_shared<CrossExecForwarderCallback>(relayExec_, fwd, std::move(relayAdapter));
467+
fwd->setCallback(std::make_shared<LocalForwarderCallback>(
474468
tlForwarders_.get(),
475469
ftn,
476470
std::move(crossExec),
477471
/*removeOnEmpty=*/false
478472
));
479473

480-
return localPubFwd;
474+
return tlForwarders_->replace(ftn, fwd);
481475
}
482476

483477
// Called from LocalPublishFilter::publish() on publisherExec. Creates the publisher's
@@ -495,12 +489,8 @@ Subscriber::PublishResult MoqxRelay::publishFromPublisherExec(
495489
tlForwarders_.reset(new LocalForwarderRegistry());
496490
}
497491

498-
auto localPubFwd = createPublisherForwarder(pub);
499-
500-
// The publisher's forwarder is authoritative — claim the entry, displacing any
501-
// stale subscribe-path local forwarder so same-thread subscribers reuse THIS
502-
// forwarder via the fast path.
503-
tlForwarders_->replace(pub.fullTrackName, localPubFwd)
492+
auto localPubFwd = std::make_shared<MoQForwarder>(pub.fullTrackName);
493+
installPublisherForwarder(pub.fullTrackName, localPubFwd)
504494
.markReady(InitialTrackState{pub.largest, pub.extensions});
505495

506496
// crossExecFilter is a channel subscriber for the relay exec

src/MoqxRelay.h

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -363,9 +363,10 @@ class MoqxRelay : public moxygen::Publisher,
363363
std::shared_ptr<moxygen::MoQSession> session
364364
);
365365

366-
// Constructs the publisher's local forwarder and installs its callback chain on
367-
// publisherExec. tlForwarders_ must already be initialized.
368-
std::shared_ptr<moxygen::MoQForwarder> createPublisherForwarder(const moxygen::PublishRequest& pub
366+
// Runs on fwd's exec; the returned Claim owes a markReady/fail.
367+
LocalForwarderRegistry::Claim installPublisherForwarder(
368+
const moxygen::FullTrackName& ftn,
369+
const std::shared_ptr<moxygen::MoQForwarder>& fwd
369370
);
370371

371372
std::optional<moxygen::PublishError>

0 commit comments

Comments
 (0)