Skeleton for MoQTestClient with main method

Summary: Structure for MoqTestClient class with functions defined and main method.

Reviewed By: sharmafb

Differential Revision: D75456552

fbshipit-source-id: 6eea901eb94a5e0f9b0872ae9931c43e94e92ea2

Author: Sujay Patel

moxygen/MoQSession.cpp

@@ -3162,7 +3162,6 @@ folly::coro::Task<Publisher::SubscribeResult> MoQSession::subscribe(
   auto subTrack = subTracks_.try_emplace(trackAlias, trackReceiveState);
   XCHECK(subTrack.second) << "Track alias already in use alias=" << trackAlias
                           << " sess=" << this;
-
   auto subscribeResult = co_await trackReceiveState->subscribeFuture();
   XLOG(DBG1) << "Subscribe ready trackReceiveState=" << trackReceiveState
              << " requestID=" << reqID;

moxygen/moqtest/MoQTestClient.cpp

@@ -0,0 +1,135 @@
+// (c) Meta Platforms, Inc. and affiliates. Confidential and proprietary.
+
+#include "moxygen/moqtest/MoQTestClient.h"
+#include "folly/coro/BlockingWait.h"
+#include "folly/init/Init.h"
+#include "folly/io/async/ScopedEventBaseThread.h"
+#include "moxygen/MoQClient.h"
+#include "moxygen/moqtest/Utils.h"
+
+namespace moxygen {
+
+DEFINE_string(url, "http://localhost:9999", "URL to connect to");
+DEFINE_int32(connect_timeout, 1000, "connect timeout in ms");
+DEFINE_int32(transaction_timeout, 1000, "transaction timeout in ms");
+const int kDefaultRequestId = 0;
+const std::string kDefaultTrackName = "test";
+const GroupOrder kDefaultGroupOrder = GroupOrder(0x0);
+const LocationType kDefaultLocationType = LocationType(1);
+const uint64_t kDefaultEndGroup = 128;
+
+MoQTestClient::MoQTestClient(folly::EventBase* evb, proxygen::URL url)
+    : moqClient_(std::make_unique<MoQClient>(evb, std::move(url))) {}
+
+folly::coro::Task<void> MoQTestClient::connect(folly::EventBase* evb) {
+  co_await moqClient_->setupMoQSession(
+      std::chrono::milliseconds(FLAGS_connect_timeout),
+      std::chrono::seconds(FLAGS_transaction_timeout),
+      nullptr,
+      shared_from_this());
+  co_return;
+}
+
+void MoQTestClient::initialize() {
+  // Create a receiver for the client
+  subReceiver_ = std::make_shared<ObjectReceiver>(
+      ObjectReceiver::SUBSCRIBE,
+      std::shared_ptr<MoQTestClient>(shared_from_this()));
+  fetchReceiver_ = std::make_shared<ObjectReceiver>(
+      ObjectReceiver::FETCH,
+      std::shared_ptr<MoQTestClient>(shared_from_this()));
+}
+
+folly::coro::Task<moxygen::TrackNamespace> MoQTestClient::subscribe(
+    MoQTestParameters params) {
+  auto trackNamespace = convertMoqTestParamToTrackNamespace(&params);
+
+  // Create a SubRequest with the created TrackNamespace as its fullTrackName
+  SubscribeRequest sub;
+  sub.requestID = kDefaultRequestId;
+
+  FullTrackName ftn;
+  ftn.trackNamespace = trackNamespace.value();
+  ftn.trackName = kDefaultTrackName;
+
+  sub.fullTrackName = ftn;
+  sub.trackAlias = TrackAlias();
+  sub.groupOrder = kDefaultGroupOrder;
+  sub.locType = kDefaultLocationType;
+  sub.endGroup = kDefaultEndGroup;
+
+  // Subscribe to the reciever
+  auto res = co_await moqClient_->moqSession_->subscribe(sub, subReceiver_);
+
+  co_return trackNamespace.value();
+}
+
+folly::coro::Task<moxygen::TrackNamespace> MoQTestClient::fetch(
+    MoQTestParameters params) {
+  auto trackNamespace = convertMoqTestParamToTrackNamespace(&params);
+
+  // Create a Fetch with the created TrackNamespace as its fullTrackName
+  Fetch fetch;
+  fetch.requestID = kDefaultRequestId;
+
+  FullTrackName ftn;
+  ftn.trackNamespace = trackNamespace.value();
+  ftn.trackName = kDefaultTrackName;
+  fetch.fullTrackName = ftn;
+  fetch.groupOrder = kDefaultGroupOrder;
+
+  // Fetch to the reciever
+  auto res = co_await moqClient_->moqSession_->fetch(fetch, fetchReceiver_);
+
+  co_return trackNamespace.value();
+}
+
+ObjectReceiverCallback::FlowControlState MoQTestClient::onObject(
+    const ObjectHeader& objHeader,
+    Payload payload) {
+  std::cout << "onObject" << std::endl;
+  return ObjectReceiverCallback::FlowControlState::BLOCKED;
+}
+
+void MoQTestClient::onObjectStatus(const ObjectHeader& objHeader) {
+  std::cout << "onObjectStatus" << std::endl;
+}
+
+void MoQTestClient::onEndOfStream() {
+  std::cout << "onEndOfStream" << std::endl;
+}
+
+void MoQTestClient::onError(ResetStreamErrorCode) {
+  std::cout << "onError" << std::endl;
+}
+void MoQTestClient::onSubscribeDone(SubscribeDone done) {
+  std::cout << "onSubscribeDone" << std::endl;
+}
+
+} // namespace moxygen
+
+int main(int argc, char** argv) {
+  gflags::ParseCommandLineFlags(&argc, &argv, false);
+  folly::Init init(&argc, &argv);
+
+  folly::ScopedEventBaseThread evb;
+
+  // Initialize Client with url
+  moxygen::MoQTestParameters kDefaultMoqParams;
+  auto url = proxygen::URL(moxygen::FLAGS_url);
+  std::shared_ptr<moxygen::MoQTestClient> client =
+      std::make_shared<moxygen::MoQTestClient>(evb.getEventBase(), url);
+  client->initialize();
+
+  // Connect Client to Server
+  folly::coro::blockingWait(
+      client->connect(evb.getEventBase()).scheduleOn(evb.getEventBase()));
+
+  // Test a Subscribe Call
+  folly::coro::blockingWait(
+      client->subscribe(kDefaultMoqParams).scheduleOn(evb.getEventBase()));
+
+  sleep(10);
+
+  return 0;
+}

moxygen/moqtest/MoQTestClient.h

@@ -0,0 +1,43 @@
+// (c) Meta Platforms, Inc. and affiliates. Confidential and proprietary.
+
+#pragma once
+
+#include "moxygen/MoQClient.h"
+#include "moxygen/ObjectReceiver.h"
+#include "moxygen/Subscriber.h"
+#include "moxygen/moqtest/Types.h"
+
+namespace moxygen {
+
+class MoQTestClient : public moxygen::Subscriber,
+                      public std::enable_shared_from_this<MoQTestClient>,
+                      public ObjectReceiverCallback {
+ public:
+  MoQTestClient(folly::EventBase* evb, proxygen::URL url);
+
+  ~MoQTestClient() override {}
+
+  folly::coro::Task<void> connect(folly::EventBase* evb);
+
+  void initialize();
+
+  folly::coro::Task<moxygen::TrackNamespace> subscribe(
+      MoQTestParameters params);
+
+  folly::coro::Task<moxygen::TrackNamespace> fetch(MoQTestParameters params);
+
+  // Override Vritual Functions for now to return basic print statements
+  virtual FlowControlState onObject(
+      const ObjectHeader& objHeader,
+      Payload payload) override;
+  virtual void onObjectStatus(const ObjectHeader& objHeader) override;
+  virtual void onEndOfStream() override;
+  virtual void onError(ResetStreamErrorCode) override;
+  virtual void onSubscribeDone(SubscribeDone done) override;
+
+ private:
+  std::unique_ptr<MoQClient> moqClient_;
+  std::shared_ptr<ObjectReceiver> subReceiver_;
+  std::shared_ptr<ObjectReceiver> fetchReceiver_;
+};
+} // namespace moxygen