quiche/quic/moqt/moqt_fetch_task.h

// Copyright 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_MOQT_FETCH_TASK_H_
#define QUICHE_QUIC_MOQT_MOQT_FETCH_TASK_H_

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

#include "absl/base/nullability.h"
#include "absl/status/status.h"
#include "quiche/quic/moqt/moqt_error.h"
#include "quiche/quic/moqt/moqt_key_value_pair.h"
#include "quiche/quic/moqt/moqt_names.h"
#include "quiche/quic/moqt/moqt_object.h"
#include "quiche/common/quiche_callbacks.h"
#include "quiche/web_transport/web_transport.h"

namespace moqt {

// The callback we'll use for most request types going forward. Can only be used
// once. If the argument is MessageParameters, the request was successful.
// Otherwise, the request failed with the given error info.
// This is not used for responses to SUBSCRIBE (SubscribeVisitor), FETCH
// (FetchResponseCallback), and TRACK_STATUS (not implemented, but includes
// Track Properties).
using MoqtResponseCallback = quiche::SingleUseCallback<void(
    std::variant<MessageParameters, MoqtRequestErrorInfo>)>;

// TODO(martinduke): There are will be multiple instances of flow-controlled
// "pull" data retrieval tasks. It might be worthwhile to extract some common
// features into a base class.

using ObjectsAvailableCallback = quiche::MultiUseCallback<void()>;
// Potential results of a GetNextObject/GetNextMessage() call.
enum GetNextResult {
  // The next object or message is available, and is placed into the reference
  // specified by the caller.
  kSuccess,
  // The next object or message is not yet available (equivalent of EAGAIN).
  kPending,
  // The end of the response has been reached.
  kEof,
  // The request has failed; the error is available.
  kError,
};

enum class TransactionType : uint8_t { kAdd, kDelete };

// A handle representing a fetch in progress.  The fetch in question can be
// cancelled by deleting the object.
class MoqtFetchTask {
 public:
  virtual ~MoqtFetchTask() = default;

  // TODO(martinduke): Replace with GetNextResult above.
  // Potential results of a GetNextObject() call.
  enum GetNextObjectResult {
    // The next object is available, and is placed into the reference specified
    // by the caller.
    kSuccess,
    // The next object is not yet available (equivalent of EAGAIN).
    kPending,
    // The end of fetch has been reached.
    kEof,
    // The fetch has failed; the error is available via GetStatus().
    kError,
  };

  // Returns the next object received via the fetch, if available.
  virtual GetNextObjectResult GetNextObject(PublishedObject& output) = 0;

  // Sets the callback that is called when GetNextObject() has previously
  // returned kPending, but now a new object (or potentially an error or an
  // end-of-fetch) is available. The application is responsible for calling
  // GetNextObject() until it gets kPending; no further callback will occur
  // until then.
  // If an object is available immediately, the callback will be called
  // immediately.
  virtual void SetObjectAvailableCallback(
      ObjectsAvailableCallback callback) = 0;

  // Returns the error if fetch has completely failed, and OK otherwise.
  virtual absl::Status GetStatus() = 0;
};

class MoqtNamespaceTask {
 public:
  virtual ~MoqtNamespaceTask() = default;

  // The provided callback may be immediately invoked.
  virtual void SetObjectsAvailableCallback(ObjectsAvailableCallback
                                           absl_nullable callback) = 0;

  // Returns the state of the message queue. If available, writes the suffix
  // into |suffix|. If |type| is kAdd, it is from a NAMESPACE message. If |type|
  // is kDelete, it is from a NAMESPACE_DONE message.
  virtual GetNextResult GetNextSuffix(TrackNamespace& suffix,
                                      TransactionType& type) = 0;

  // Returns the error if request has completely failed, and nullopt otherwise.
  virtual std::optional<webtransport::StreamErrorCode> GetStatus() = 0;

  // Handle a REQUEST_UPDATE message.
  virtual void Update(const MessageParameters& parameters,
                      MoqtResponseCallback response_callback) = 0;

  // Returns the prefix for this task.
  virtual const TrackNamespace& prefix() = 0;
};

}  // namespace moqt

#endif  // QUICHE_QUIC_MOQT_MOQT_FETCH_TASK_H_