Full Publisher For Forwarding Preference One

Summary: Fully implemented Subscribe call for MoQTestServer for TrackNamespaces that have a Forwarding Preference of 1.

Reviewed By: sharmafb

Differential Revision: D75614231

fbshipit-source-id: c1bba81dccd9bc7a29b467b4da8f934b2d1a9875

Author: Sujay Patel

moxygen/moqtest/MoQTestServer.cpp

@@ -70,7 +70,13 @@ folly::coro::Task<void> MoQTestServer::onSubscribe(
   // Switch based on forwarding preference
   switch (params.forwardingPreference) {
     case (ForwardingPreference::ONE_SUBGROUP_PER_GROUP): {
-      co_await sendObjectsForForwardPreferenceZero(params, callback);
+      co_await sendOneSubgroupPerGroup(params, callback);
+      break;
+    }
+
+    case (ForwardingPreference::ONE_SUBGROUP_PER_OBJECT): {
+      co_await sendOneSubgroupPerObject(params, callback);
+
       break;
     }
 
@@ -89,7 +95,7 @@ folly::coro::Task<void> MoQTestServer::onSubscribe(
   co_return;
 }
 
-folly::coro::Task<void> MoQTestServer::sendObjectsForForwardPreferenceZero(
+folly::coro::Task<void> MoQTestServer::sendOneSubgroupPerGroup(
     MoQTestParameters params,
     std::shared_ptr<TrackConsumer> callback) {
   // Iterate through Groups
@@ -136,4 +142,51 @@ folly::coro::Task<void> MoQTestServer::sendObjectsForForwardPreferenceZero(
   }
 }
 
+folly::coro::Task<void> MoQTestServer::sendOneSubgroupPerObject(
+    MoQTestParameters params,
+    std::shared_ptr<TrackConsumer> callback) {
+  // Iterate through Objects
+  for (int groupNum = params.startGroup; groupNum <= params.lastGroupInTrack;
+       groupNum += params.groupIncrement) {
+    // Iterate Through Objects in SubGroup
+    for (int objectId = params.startObject;
+         objectId <= params.lastObjectInTrack;
+         objectId += params.objectIncrement) {
+      // Begin a New Subgroup per object (Default Priority)
+      auto maybeSubConsumer =
+          callback->beginSubgroup(groupNum, objectId, kDefaultPriority);
+      auto subConsumer = maybeSubConsumer->get();
+      // Find Object Size
+      int objectSize = moxygen::getObjectSize(objectId, &params);
+
+      // Add Integer/Variable Extensions if needed
+      std::vector<Extension> extensions = getExtensions(
+          params.testIntegerExtension, params.testVariableExtension);
+
+      // If there are send end of group markers and j == lastObjectID, send
+      // the end of group
+      if (objectId < params.lastObjectInTrack ||
+          !params.sendEndOfGroupMarkers) {
+        // Begin Delivering Object With Payload
+        std::string p = std::string(objectSize, 't');
+        auto objectPayload = folly::IOBuf::copyBuffer(p);
+        subConsumer->object(
+            objectId, std::move(objectPayload), extensions, false);
+      } else {
+        subConsumer->endOfGroup(objectId);
+      }
+
+      // If SubGroup Hasn't Been Ended Already
+      if (!params.sendEndOfGroupMarkers) {
+        subConsumer->endOfSubgroup();
+      }
+
+      // Set Delay Based on Object Frequency
+      co_await folly::coro::sleep(
+          std::chrono::milliseconds(params.objectFrequency));
+    }
+  }
+  co_return;
+}
+
 } // namespace moxygen

moxygen/moqtest/MoQTestServer.h

@@ -39,7 +39,11 @@ class MoQTestServer : public moxygen::Publisher,
       SubscribeRequest sub,
       std::shared_ptr<TrackConsumer> callback);
 
-  folly::coro::Task<void> sendObjectsForForwardPreferenceZero(
+  folly::coro::Task<void> sendOneSubgroupPerGroup(
+      MoQTestParameters params,
+      std::shared_ptr<TrackConsumer> callback);
+
+  folly::coro::Task<void> sendOneSubgroupPerObject(
       MoQTestParameters params,
       std::shared_ptr<TrackConsumer> callback);
 

moxygen/test/MoQTrackServerTest.cpp

@@ -59,7 +59,7 @@ class MoQTrackServerTest : public testing::Test {
 
 } // namespace
 
-// Subscription Testing - ForwardPreference=0
+// Subscription Testing
 TEST_F(MoQTrackServerTest, ValidateSubscribeWithForwardPreferenceZero) {
   MoQTrackServerTest::CreateDefaultMoQTestParameters();
 
@@ -102,8 +102,7 @@ TEST_F(MoQTrackServerTest, ValidateSubscribeWithForwardPreferenceZero) {
   }
 
   // Call the onSubscribe method
-  auto task =
-      server_.sendObjectsForForwardPreferenceZero(params_, mockConsumer);
+  auto task = server_.sendOneSubgroupPerGroup(params_, mockConsumer);
 
   // Wait for the coroutine to complete
   folly::coro::blockingWait(std::move(task));
@@ -176,8 +175,57 @@ TEST_F(
   }
 
   // Call the onSubscribe method
-  auto task =
-      server_.sendObjectsForForwardPreferenceZero(params_, mockConsumer);
+  auto task = server_.sendOneSubgroupPerGroup(params_, mockConsumer);
+
+  // Wait for the coroutine to complete
+  folly::coro::blockingWait(std::move(task));
+}
+
+TEST_F(MoQTrackServerTest, ValidateSubscribeWithForwardPreferenceOne) {
+  MoQTrackServerTest::CreateDefaultMoQTestParameters();
+  params_.forwardingPreference = moxygen::ForwardingPreference(1);
+
+  // Create a mock track consumer
+  auto mockConsumer = std::make_shared<moxygen::MockTrackConsumer>();
+
+  // Create a mock subgroup consumer
+  auto mockSubgroupConsumer = std::make_shared<moxygen::MockSubgroupConsumer>();
+
+  // Set expectations for beginSubgroup
+  for (int groupId = 0; groupId <= 10; groupId++) {
+    for (int objectId = 0; objectId <= params_.lastObjectInTrack; objectId++) {
+      // Set expectations for beginObject
+      int objectSize = moxygen::getObjectSize(objectId, &params_);
+      // Create a mock subgroup consumer
+      auto mockSubgroupConsumer =
+          std::make_shared<moxygen::MockSubgroupConsumer>();
+      EXPECT_CALL(*mockConsumer, beginSubgroup(groupId, objectId, testing::_))
+          .Times(1)
+          .WillOnce(testing::Return(mockSubgroupConsumer));
+      EXPECT_CALL(
+          *mockSubgroupConsumer,
+          object(objectId, testing::_, testing::_, testing::_))
+          .Times(1)
+          .WillOnce(testing::Invoke(
+              [objectSize](
+                  auto, std::unique_ptr<folly::IOBuf> payload, auto, auto) {
+                int payloadLength = (*payload).length();
+                EXPECT_EQ(payloadLength, objectSize);
+                return folly::Expected<folly::Unit, moxygen::MoQPublishError>(
+                    {});
+              }))
+          .WillOnce(::testing::Return(
+              folly::Expected<folly::Unit, moxygen::MoQPublishError>({})));
+      // Set expectations for endOfSubgroup
+      EXPECT_CALL(*mockSubgroupConsumer, endOfSubgroup())
+          .Times(1)
+          .WillOnce(::testing::Return(
+              folly::Expected<folly::Unit, moxygen::MoQPublishError>({})));
+    }
+  }
+
+  // Call the onSubscribe method
+  auto task = server_.sendOneSubgroupPerObject(params_, mockConsumer);
 
   // Wait for the coroutine to complete
   folly::coro::blockingWait(std::move(task));