quiche/quic/moqt/tools/chat_client.h

// Copyright (c) 2024 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#ifndef QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H
#define QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H

#include <cstdint>
#include <memory>
#include <optional>
#include <variant>

#include "absl/base/nullability.h"
#include "absl/container/flat_hash_set.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/core/io/quic_event_loop.h"
#include "quiche/quic/core/quic_server_id.h"
#include "quiche/quic/core/quic_time.h"
#include "quiche/quic/moqt/moqt_error.h"
#include "quiche/quic/moqt/moqt_fetch_task.h"
#include "quiche/quic/moqt/moqt_key_value_pair.h"
#include "quiche/quic/moqt/moqt_known_track_publisher.h"
#include "quiche/quic/moqt/moqt_messages.h"
#include "quiche/quic/moqt/moqt_names.h"
#include "quiche/quic/moqt/moqt_object.h"
#include "quiche/quic/moqt/moqt_outgoing_queue.h"
#include "quiche/quic/moqt/moqt_session.h"
#include "quiche/quic/moqt/moqt_session_callbacks.h"
#include "quiche/quic/moqt/moqt_session_interface.h"
#include "quiche/quic/moqt/tools/moqt_client.h"
#include "quiche/common/platform/api/quiche_export.h"
#include "quiche/common/quiche_callbacks.h"

namespace moqt::moq_chat {

constexpr quic::QuicTime::Delta kChatEventLoopDuration =
    quic::QuicTime::Delta::FromMilliseconds(500);

// Chat clients accept a ChatUserInterface that implements how user input is
// captured, and peer messages are displayed.
class ChatUserInterface {
 public:
  virtual ~ChatUserInterface() = default;

  // ChatUserInterface cannot be used until initialized. This is separate from
  // the constructor, because the constructor might create the event loop.
  // |callback| is what ChatUserInterface will call when there is user input.
  // |event_loop| is the event loop that the ChatUserInterface should use.
  virtual void Initialize(
      quiche::MultiUseCallback<void(absl::string_view)> callback,
      quic::QuicEventLoop* event_loop) = 0;
  // Write a peer message to the user output.
  virtual void WriteToOutput(absl::string_view user,
                             absl::string_view message) = 0;
  // Run the event loop for a short interval and exit.
  virtual void IoLoop() = 0;
};

class ChatClient {
 public:
  // If |event_loop| is nullptr, a new one will be created. If multiple
  // endpoints are running on the same thread, as in tests, they should share
  // an event loop.
  ChatClient(const quic::QuicServerId& server_id, bool ignore_certificate,
             std::unique_ptr<ChatUserInterface> interface,
             absl::string_view chat_id, absl::string_view username,
             absl::string_view localhost,
             quic::QuicEventLoop* event_loop = nullptr);
  ~ChatClient() {
    if (!session_is_open_) {
      return;
    }
    session_is_open_ = false;
    if (session_ != nullptr) {
      // Closing the session can trigger a number of callbacks. The application
      // is tearing down too, so negate them.r;
      session_->Close();
      session_ = nullptr;
    }
  }

  // Establish the MoQT session. Returns false if it fails.
  bool Connect(absl::string_view path);

  void OnTerminalLineInput(absl::string_view input_message);

  // Run the event loop until an input or output event is ready, or the
  // session closes.
  void IoLoop() {
    while (session_is_open_) {
      interface_->IoLoop();
    }
  }

  void WriteToOutput(absl::string_view user, absl::string_view message) {
    if (interface_ != nullptr) {
      interface_->WriteToOutput(user, message);
    }
  }

  quic::QuicEventLoop* event_loop() { return event_loop_; }

  class QUICHE_EXPORT ObjectSubscriberVisitor : public moqt::SubscribeVisitor {
   public:
    ObjectSubscriberVisitor(ChatClient* client) : client_(client) {}

    void OnReply(
        const moqt::FullTrackName& full_track_name,
        std::variant<SubscribeOkData, MoqtRequestErrorInfo> response) override;

    void OnCanAckObjects(MoqtObjectAckFunction) override {}

    void OnObjectFragment(const moqt::FullTrackName& full_track_name,
                          const PublishedObjectMetadata& metadata,
                          absl::string_view object, uint64_t offset) override;

    void OnPublishDone(FullTrackName) override {}

    // TODO(martinduke): Implement this.
    void OnMalformedTrack(const FullTrackName& /*full_track_name*/) override {}

    void OnStreamFin(const FullTrackName&, DataStreamIndex) override {}
    void OnStreamReset(const FullTrackName&, DataStreamIndex) override {}

   private:
    ChatClient* client_;
  };

  // Returns false on error.
  bool PublishNamespaceAndSubscribeNamespace();

  bool session_is_open() const { return session_is_open_; }

  // Returns true if the client has outstanding subscribes.
  bool is_syncing() const { return subscribes_to_make_ > 0; }

 private:
  void RunEventLoop() { event_loop_->RunEventLoopOnce(kChatEventLoopDuration); }
  // Callback for incoming publish_namespaces. If |parameters| is not nullopt,
  // it's a PUBLISH_NAMESPACE or NAMESPACE. If |callback| is not nullptr, it's
  // a PUBLISH_NAMESPACE.
  void OnIncomingPublishNamespace(
      const moqt::TrackNamespace& track_namespace,
      const MessageParameters* absl_nullable parameters,
      moqt::MoqtResponseCallback absl_nullable callback);

  // Basic session information
  FullTrackName my_track_name_;

  // Related to subscriptions/publish_namespaces
  // TODO: One for each subscribe
  ObjectSubscriberVisitor remote_track_visitor_;

  // General state variables
  // The event loop to use for this client.
  quic::QuicEventLoop* event_loop_;
  // If the client created its own event loop, it will own it.
  std::unique_ptr<quic::QuicEventLoop> local_event_loop_;
  bool connect_failed_ = false;
  bool session_is_open_ = false;
  moqt::MoqtSession* session_ = nullptr;
  moqt::MoqtKnownTrackPublisher publisher_;
  std::unique_ptr<moqt::MoqtClient> client_;
  moqt::MoqtSessionCallbacks session_callbacks_;
  std::unique_ptr<moqt::MoqtNamespaceTask> namespace_task_;

  // Related to syncing.
  absl::flat_hash_set<FullTrackName> other_users_;
  int subscribes_to_make_ = 0;

  // Handling outgoing messages
  std::shared_ptr<moqt::MoqtOutgoingQueue> queue_;

  // User interface for input and output.
  std::unique_ptr<ChatUserInterface> interface_;
};

}  // namespace moqt::moq_chat

#endif  // QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H