Skip to content

Run format.sh #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
afrind wants to merge 1 commit into from
Closed

Run format.sh #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 15 additions & 12 deletions include/o_rly/ORelay.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,10 @@ class ORelay : public moxygen::Publisher,
std::shared_ptr<NamespacePublishHandle> namespacePublishHandle
) override;

folly::coro::Task<moxygen::Subscriber::PublishNamespaceResult>
publishNamespace(moxygen::PublishNamespace ann, std::shared_ptr<moxygen::Subscriber::PublishNamespaceCallback>)
override;
folly::coro::Task<moxygen::Subscriber::PublishNamespaceResult> publishNamespace(
moxygen::PublishNamespace ann,
std::shared_ptr<moxygen::Subscriber::PublishNamespaceCallback>
) override;

PublishResult publish(
moxygen::PublishRequest pubReq,
Expand All @@ -61,8 +62,8 @@ class ORelay : public moxygen::Publisher,
XLOG(INFO) << "Processing goaway uri=" << goaway.newSessionUri;
}

std::shared_ptr<moxygen::MoQSession> findPublishNamespaceSession(const moxygen::TrackNamespace& ns
);
std::shared_ptr<moxygen::MoQSession>
findPublishNamespaceSession(const moxygen::TrackNamespace& ns);

// Wrapper for compatibility - returns single session as vector
std::vector<std::shared_ptr<moxygen::MoQSession>>
Expand Down Expand Up @@ -98,13 +99,15 @@ class ORelay : public moxygen::Publisher,

void publishNamespaceDone() override { relay_.publishNamespaceDone(trackNamespace_, this); }

folly::coro::Task<RequestUpdateResult> requestUpdate(moxygen::RequestUpdate reqUpdate
) override {
co_return folly::makeUnexpected(moxygen::RequestError{
reqUpdate.requestID,
moxygen::RequestErrorCode::NOT_SUPPORTED,
"REQUEST_UPDATE not supported for relay PUBLISH_NAMESPACE"
});
folly::coro::Task<RequestUpdateResult>
requestUpdate(moxygen::RequestUpdate reqUpdate) override {
co_return folly::makeUnexpected(
moxygen::RequestError{
reqUpdate.requestID,
moxygen::RequestErrorCode::NOT_SUPPORTED,
"REQUEST_UPDATE not supported for relay PUBLISH_NAMESPACE"
}
);
}

// Helper to check if THIS node (excluding children) has content
Expand Down
72 changes: 42 additions & 30 deletions src/ORelay.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,11 +81,13 @@ folly::coro::Task<Subscriber::PublishNamespaceResult> ORelay::publishNamespace(
XLOG(DBG1) << __func__ << " ns=" << ann.trackNamespace;
// check auth
if (!ann.trackNamespace.startsWith(allowedNamespacePrefix_)) {
co_return folly::makeUnexpected(PublishNamespaceError{
ann.requestID,
PublishNamespaceErrorCode::UNINTERESTED,
"bad namespace"
});
co_return folly::makeUnexpected(
PublishNamespaceError{
ann.requestID,
PublishNamespaceErrorCode::UNINTERESTED,
"bad namespace"
}
);
}
std::vector<std::pair<std::shared_ptr<MoQSession>, bool>> sessions;
auto nodePtr = findNamespaceNode(
Expand Down Expand Up @@ -476,11 +478,13 @@ class ORelay::NamespaceSubscription : public Publisher::SubscribeNamespaceHandle
}

folly::coro::Task<RequestUpdateResult> requestUpdate(RequestUpdate reqUpdate) override {
co_return folly::makeUnexpected(RequestError{
reqUpdate.requestID,
RequestErrorCode::NOT_SUPPORTED,
"REQUEST_UPDATE not supported for relay SUBSCRIBE_NAMESPACE"
});
co_return folly::makeUnexpected(
RequestError{
reqUpdate.requestID,
RequestErrorCode::NOT_SUPPORTED,
"REQUEST_UPDATE not supported for relay SUBSCRIBE_NAMESPACE"
}
);
}

private:
Expand Down Expand Up @@ -532,11 +536,13 @@ folly::coro::Task<Publisher::SubscribeNamespaceResult> ORelay::subscribeNamespac
XLOG(DBG1) << __func__ << " nsp=" << subNs.trackNamespacePrefix;
// check auth
if (subNs.trackNamespacePrefix.empty()) {
co_return folly::makeUnexpected(SubscribeNamespaceError{
subNs.requestID,
SubscribeNamespaceErrorCode::NAMESPACE_PREFIX_UNKNOWN,
"empty"
});
co_return folly::makeUnexpected(
SubscribeNamespaceError{
subNs.requestID,
SubscribeNamespaceErrorCode::NAMESPACE_PREFIX_UNKNOWN,
"empty"
}
);
}
auto session = MoQSession::getRequestSession();
auto nodePtr = findNamespaceNode(subNs.trackNamespacePrefix, /*createMissingNodes=*/true);
Expand Down Expand Up @@ -718,11 +724,13 @@ ORelay::subscribe(SubscribeRequest subReq, std::shared_ptr<TrackConsumer> consum
if (!subscriber) {
XLOG(ERR) << "addSubscriber returned null (draining?) for " << subReq.fullTrackName
<< " reqID=" << subReq.requestID;
co_return folly::makeUnexpected(SubscribeError{
subReq.requestID,
SubscribeErrorCode::INTERNAL_ERROR,
"failed to add subscriber"
});
co_return folly::makeUnexpected(
SubscribeError{
subReq.requestID,
SubscribeErrorCode::INTERNAL_ERROR,
"failed to add subscriber"
}
);
}
XLOG(DBG4) << "added subscriber for ftn=" << subReq.fullTrackName;
// As per the spec, we must set forward = true in the subscribe request
Expand Down Expand Up @@ -788,11 +796,13 @@ ORelay::subscribe(SubscribeRequest subReq, std::shared_ptr<TrackConsumer> consum
auto& forwarder = subscriptionIt->second.forwarder;
if (forwarder->largest() && subReq.locType == LocationType::AbsoluteRange &&
subReq.endGroup < forwarder->largest()->group) {
co_return folly::makeUnexpected(SubscribeError{
subReq.requestID,
SubscribeErrorCode::INVALID_RANGE,
"Range in the past, use FETCH"
});
co_return folly::makeUnexpected(
SubscribeError{
subReq.requestID,
SubscribeErrorCode::INVALID_RANGE,
"Range in the past, use FETCH"
}
);
// start may be in the past, it will get adjusted forward to largest
}
bool forwarding = subscriptionIt->second.forwarder->numForwardingSubscribers() > 0;
Expand All @@ -801,11 +811,13 @@ ORelay::subscribe(SubscribeRequest subReq, std::shared_ptr<TrackConsumer> consum
if (!subscriber) {
XLOG(ERR) << "addSubscriber returned null (draining?) for " << subReq.fullTrackName
<< " reqID=" << subReq.requestID;
co_return folly::makeUnexpected(SubscribeError{
subReq.requestID,
SubscribeErrorCode::INTERNAL_ERROR,
"failed to add subscriber"
});
co_return folly::makeUnexpected(
SubscribeError{
subReq.requestID,
SubscribeErrorCode::INTERNAL_ERROR,
"failed to add subscriber"
}
);
}
XLOG(DBG4) << "added subscriber for ftn=" << subReq.fullTrackName;
if (!forwarding && subscriptionIt->second.forwarder->numForwardingSubscribers() > 0) {
Expand Down
3 changes: 2 additions & 1 deletion src/config/loader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@ ParsedConfig loadConfig(const std::string& path, bool strict) {

std::string generateSchema() {
return rfl::json::to_schema<
rfl::Description<"Configuration schema for the o-rly relay.", ParsedConfig>>(rfl::json::pretty
rfl::Description<"Configuration schema for the o-rly relay.", ParsedConfig>>(
rfl::json::pretty
);
}

Expand Down
93 changes: 54 additions & 39 deletions tests/ORelayTest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,8 +299,9 @@ class ORelayTest : public ::testing::Test {
ON_CALL(*sg, beginObject(_, _, _, _))
.WillByDefault(Return(folly::makeExpected<MoQPublishError>(folly::unit)));
ON_CALL(*sg, objectPayload(_, _))
.WillByDefault(Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS)
));
.WillByDefault(
Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS))
);
ON_CALL(*sg, endOfSubgroup())
.WillByDefault(Return(folly::makeExpected<MoQPublishError>(folly::unit)));
ON_CALL(*sg, endOfGroup(_))
Expand Down Expand Up @@ -481,7 +482,8 @@ TEST_F(ORelayTest, PruneOnPublishDoneBug) {
EXPECT_EQ(state.session, nullptr); // No session - PASS

// THIS FAILS: Node should have been pruned but still exists (memory leak)
EXPECT_FALSE(state.nodeExists
EXPECT_FALSE(
state.nodeExists
) << "BUG: Node test/A/B/C still exists after publish ended and was the "
"only content. "
"onPublishDone should have called tryPruneChild to clean up empty "
Expand Down Expand Up @@ -791,8 +793,9 @@ TEST_F(ORelayTest, ForwarderOnlyCreatesSubgroupsBeforeObjectData) {
ON_CALL(*sg, beginObject(_, _, _, _))
.WillByDefault(Return(folly::makeExpected<MoQPublishError>(folly::unit)));
ON_CALL(*sg, objectPayload(_, _))
.WillByDefault(Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS)
));
.WillByDefault(
Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS))
);
ON_CALL(*sg, endOfSubgroup())
.WillByDefault(Return(folly::makeExpected<MoQPublishError>(folly::unit)));
};
Expand Down Expand Up @@ -878,7 +881,8 @@ TEST_F(ORelayTest, GracefulSessionDraining) {
EXPECT_CALL(*consumers[0], beginSubgroup(i, 0, _, _))
.WillOnce([this, i, &sub0_sgs](uint64_t, uint64_t, uint8_t, bool) {
sub0_sgs[i] = createMockSubgroupConsumer();
return folly::makeExpected<MoQPublishError, std::shared_ptr<SubgroupConsumer>>(sub0_sgs[i]
return folly::makeExpected<MoQPublishError, std::shared_ptr<SubgroupConsumer>>(
sub0_sgs[i]
);
});
}
Expand Down Expand Up @@ -1076,10 +1080,11 @@ TEST_F(ORelayTest, SubscriberUnsubscribeDoesNotReceiveNewObjects) {

// publisher ends subscription
EXPECT_CALL(*mockConsumer1, publishDone(testing::_));
EXPECT_TRUE(publishConsumer
->publishDone({RequestID(1), PublishDoneStatusCode::TRACK_ENDED, 0, "track ended"}
)
.hasValue());
EXPECT_TRUE(
publishConsumer
->publishDone({RequestID(1), PublishDoneStatusCode::TRACK_ENDED, 0, "track ended"})
.hasValue()
);

// Subscriber 2 joins after publishDone
subscribeToTrack(
Expand Down Expand Up @@ -1116,20 +1121,20 @@ TEST_F(ORelayTest, SubscribeNamespaceDoesntAddDrainingPublish) {
auto mockConsumer1 = createMockConsumer();
EXPECT_CALL(*subscriber1, publish(testing::_, testing::_))
.WillOnce([mockConsumer1](auto pubReq, auto subHandle) {
return Subscriber::PublishResult(Subscriber::PublishConsumerAndReplyTask{
mockConsumer1,
[]() -> folly::coro::Task<folly::Expected<PublishOk, PublishError>> {
co_return PublishOk{
/*requestID=*/RequestID(1),
/*forward=*/true,
/*subscriberPriority=*/0,
/*groupOrder=*/GroupOrder::OldestFirst,
/*locType=*/LocationType::LargestObject,
/*start=*/std::nullopt,
/*endGroup=*/std::nullopt
};
}()
});
return Subscriber::PublishResult(
Subscriber::PublishConsumerAndReplyTask{
mockConsumer1,
[]() -> folly::coro::Task<folly::Expected<PublishOk, PublishError>> {
co_return PublishOk{/*requestID=*/RequestID(1),
/*forward=*/true,
/*subscriberPriority=*/0,
/*groupOrder=*/GroupOrder::OldestFirst,
/*locType=*/LocationType::LargestObject,
/*start=*/std::nullopt,
/*endGroup=*/std::nullopt};
}()
}
);
});

EXPECT_CALL(*mockConsumer1, beginSubgroup(_, _, _, _))
Expand Down Expand Up @@ -1206,11 +1211,14 @@ TEST_F(ORelayTest, DataOperationCancelledWhenAllSubscribersFail) {
.WillOnce([this, i, &sgs](uint64_t, uint64_t, uint8_t, bool) {
sgs[i] = createMockSubgroupConsumer();
EXPECT_CALL(*sgs[i], objectPayload(_, false))
.WillOnce(Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS
)))
.WillOnce(Return(folly::makeUnexpected(
MoQPublishError(MoQPublishError::WRITE_ERROR, "write failed")
)));
.WillOnce(
Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS))
)
.WillOnce(Return(
folly::makeUnexpected(
MoQPublishError(MoQPublishError::WRITE_ERROR, "write failed")
)
));
return folly::makeExpected<MoQPublishError, std::shared_ptr<SubgroupConsumer>>(sgs[i]);
});
}
Expand Down Expand Up @@ -1268,7 +1276,8 @@ TEST_F(ORelayTest, PartialSubscriberFailureDoesNotCancelData) {
.WillOnce([this, &sgs](uint64_t, uint64_t, uint8_t, bool) {
sgs[0] = createMockSubgroupConsumer();
EXPECT_CALL(*sgs[0], objectPayload(_, false))
.WillOnce(Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS))
.WillOnce(
Return(folly::makeExpected<MoQPublishError>(ObjectPublishStatus::IN_PROGRESS))
)
.WillOnce(Return(
folly::makeUnexpected(MoQPublishError(MoQPublishError::WRITE_ERROR, "write failed"))
Expand Down Expand Up @@ -1462,9 +1471,11 @@ TEST_F(ORelayTest, TombstonedSubgroupIgnoresSubsequentObjects) {
sg = createMockSubgroupConsumer();
// First object fails with CANCELLED
EXPECT_CALL(*sg, object(0, _, _, false))
.WillOnce(Return(folly::makeUnexpected(
MoQPublishError(MoQPublishError::CANCELLED, "delivery timeout")
)));
.WillOnce(Return(
folly::makeUnexpected(
MoQPublishError(MoQPublishError::CANCELLED, "delivery timeout")
)
));
// Per API contract, error implies implicit reset - no reset() call
// After tombstoning, should NOT receive any more objects
EXPECT_CALL(*sg, object(1, _, _, _)).Times(0);
Expand Down Expand Up @@ -1574,9 +1585,11 @@ TEST_F(ORelayTest, HardErrorsRemoveSubscriber) {
.WillOnce([this, &sg](uint64_t, uint64_t, uint8_t, bool) {
sg = createMockSubgroupConsumer();
EXPECT_CALL(*sg, object(0, _, _, false))
.WillOnce(Return(folly::makeUnexpected(
MoQPublishError(MoQPublishError::WRITE_ERROR, "transport broken")
)));
.WillOnce(Return(
folly::makeUnexpected(
MoQPublishError(MoQPublishError::WRITE_ERROR, "transport broken")
)
));
// Should be reset when subscriber is removed
EXPECT_CALL(*sg, reset(_)).Times(1);
return folly::makeExpected<MoQPublishError, std::shared_ptr<SubgroupConsumer>>(sg);
Expand Down Expand Up @@ -1634,9 +1647,11 @@ TEST_F(ORelayTest, EndOfSubgroupHardErrorDoesNotCrash) {
sg = createMockSubgroupConsumer();
// Override endOfSubgroup to return a hard error (WRITE_ERROR)
EXPECT_CALL(*sg, endOfSubgroup())
.WillOnce(Return(folly::makeUnexpected(
MoQPublishError(MoQPublishError::WRITE_ERROR, "transport broken")
)));
.WillOnce(Return(
folly::makeUnexpected(
MoQPublishError(MoQPublishError::WRITE_ERROR, "transport broken")
)
));
return folly::makeExpected<MoQPublishError, std::shared_ptr<SubgroupConsumer>>(sg);
});

Expand Down