feat: enqueue streams in running player
This commit is contained in:
+9
-10
@@ -197,7 +197,7 @@ struct Player {
|
||||
PlayerStatus status = PLAYER_EMPTY;
|
||||
bool previewShown = false;
|
||||
std::array<char, 512> error{};
|
||||
std::unique_ptr<playback::ProgressReports> reporter;
|
||||
const std::unique_ptr<playback::ProgressReports> reporter = std::make_unique<playback::ProgressReports>();
|
||||
bool remoteReporting = false;
|
||||
std::thread worker;
|
||||
|
||||
@@ -1429,10 +1429,10 @@ bool player_set_progress_file(Player *player, const char *path, char *error) {
|
||||
try {
|
||||
if (!player || !path || !*path) throw std::runtime_error("A progress file path is required");
|
||||
std::lock_guard<std::mutex> lock(player->mutex);
|
||||
if (player->reporter || player->stopping.load(std::memory_order_relaxed) ||
|
||||
if (player->stopping.load(std::memory_order_relaxed) ||
|
||||
player->state.status != PLAYER_EMPTY || !player->commands.empty())
|
||||
throw std::runtime_error("Progress reporting must be initialized once, before opening media");
|
||||
player->reporter = std::make_unique<playback::ProgressReports>(path);
|
||||
player->reporter->setSingleFile(path);
|
||||
return true;
|
||||
} catch (const std::exception &failure) {
|
||||
if (error) copyError(error, failure.what());
|
||||
@@ -1441,16 +1441,15 @@ bool player_set_progress_file(Player *player, const char *path, char *error) {
|
||||
}
|
||||
return false;
|
||||
}
|
||||
bool player_register_progress_file(Player *player, const char *identity, const char *path, char *error) {
|
||||
bool player_register_progress_files(Player *player, const char *const *identities,
|
||||
const char *const *paths, int count, char *error) {
|
||||
if (error) error[0] = '\0';
|
||||
try {
|
||||
if (!player || !identity || !path) throw std::runtime_error("Queue progress identity and path are required");
|
||||
if (!player) throw std::runtime_error("A player is required for queue progress registration");
|
||||
std::lock_guard<std::mutex> lock(player->mutex);
|
||||
if (player->stopping.load(std::memory_order_relaxed) ||
|
||||
player->state.status != PLAYER_EMPTY || !player->commands.empty())
|
||||
throw std::runtime_error("Queue reports must be registered before opening media");
|
||||
if (!player->reporter) player->reporter = std::make_unique<playback::ProgressReports>();
|
||||
player->reporter->registerFile(identity, path);
|
||||
if (player->stopping.load(std::memory_order_relaxed))
|
||||
throw std::runtime_error("Cannot register queue reports while the player is stopping");
|
||||
player->reporter->registerFiles(identities, paths, count);
|
||||
return true;
|
||||
} catch (const std::exception &failure) {
|
||||
if (error) copyError(error, failure.what());
|
||||
|
||||
@@ -40,9 +40,11 @@ Player *player_create(void);
|
||||
// Opt-in reporting must be initialized before opening media. Creates a new 0600
|
||||
// file exclusively; errors never change playback state. Error buffers hold 512 bytes.
|
||||
bool player_set_progress_file(Player *player, const char *path, char *error);
|
||||
// Register immutable queue destinations before opening any media. Each path is new;
|
||||
// revisits append only to the original owned inode, and unmatched identities are not reported.
|
||||
bool player_register_progress_file(Player *player, const char *identity, const char *path, char *error);
|
||||
// Register immutable queue destinations without changing current playback. Each path is new;
|
||||
// revisits append only to the original owned inode. Conflicting identities/paths are rejected.
|
||||
// Atomic batch: all registrations succeed or none do. Strings are copied before returning.
|
||||
bool player_register_progress_files(Player *player, const char *const *identities,
|
||||
const char *const *paths, int count, char *error);
|
||||
bool player_get_progress_error(Player *player, char *error);
|
||||
// Stops the worker and drains final reporting, allowing errors to be shown before
|
||||
// destroying the handle. Idempotent; do not call from an audio callback.
|
||||
|
||||
@@ -289,32 +289,72 @@ void ProgressReporter::finish() noexcept {
|
||||
if (writer.joinable()) writer.join();
|
||||
}
|
||||
|
||||
ProgressReports::ProgressReports(const std::string &path)
|
||||
: single(std::make_unique<ProgressReporter>(path)), active(single.get()) {}
|
||||
ProgressReports::ProgressReports(const std::string &path) { setSingleFile(path); }
|
||||
|
||||
void ProgressReports::setSingleFile(const std::string &path) {
|
||||
std::lock_guard lock(mutex);
|
||||
if (single || !entries.empty()) throw std::runtime_error("Progress reporting is already initialized");
|
||||
single = std::make_unique<ProgressReporter>(path);
|
||||
singlePath = path;
|
||||
active = single.get();
|
||||
}
|
||||
|
||||
void ProgressReports::registerFile(const std::string &identity, const std::string &path) {
|
||||
const char *identities[] = {identity.c_str()}, *destinations[] = {path.c_str()};
|
||||
registerFiles(identities, destinations, 1);
|
||||
}
|
||||
|
||||
void ProgressReports::registerFiles(const char *const *identities, const char *const *destinations, int count) {
|
||||
std::lock_guard lock(mutex);
|
||||
if (single || !ProgressReporter::validMediaIdentity(identity) || entries.size() >= 4096 ||
|
||||
path.empty() || path.front() != '/' || path.find('\0') != std::string::npos ||
|
||||
entries.contains(identity))
|
||||
throw std::runtime_error("Invalid or duplicate queue progress registration");
|
||||
if (!paths.insert(path).second) throw std::runtime_error("Duplicate queue progress report path");
|
||||
entries.emplace(identity, Entry{path, nullptr});
|
||||
if (!identities || !destinations || count < 1 || count > 4096 ||
|
||||
entries.size() + count > 4096)
|
||||
throw std::runtime_error("Queue reports exceed the 4096 retained registration limit");
|
||||
std::map<std::string, Entry> additions;
|
||||
std::set<std::string> newPaths;
|
||||
for (int index = 0; index < count; ++index) {
|
||||
const auto *identity = identities[index], *path = destinations[index];
|
||||
if (!identity || !path || !ProgressReporter::validMediaIdentity(identity) ||
|
||||
!*path || *path != '/' || singleIdentity == identity || entries.contains(identity) || additions.contains(identity))
|
||||
throw std::runtime_error("Invalid or conflicting queue progress identity");
|
||||
if (singlePath == path || paths.contains(path) || !newPaths.insert(path).second)
|
||||
throw std::runtime_error("Conflicting queue progress report path");
|
||||
additions.emplace(identity, Entry{path, nullptr});
|
||||
}
|
||||
// Node merges allocate nothing: failed validation cannot partially register a batch.
|
||||
entries.merge(additions);
|
||||
paths.merge(newPaths);
|
||||
}
|
||||
|
||||
void ProgressReports::observe(const PlayerState &state) noexcept {
|
||||
std::lock_guard lock(mutex);
|
||||
if (active) active->observe(state);
|
||||
}
|
||||
|
||||
void ProgressReports::closeMedia() noexcept {
|
||||
std::lock_guard lock(mutex);
|
||||
if (active) active->closeMedia();
|
||||
}
|
||||
|
||||
void ProgressReports::begin(const std::string &identity, bool live) noexcept {
|
||||
std::lock_guard lock(mutex);
|
||||
if (single) { active->begin(identity, live); return; }
|
||||
const auto found = entries.find(identity);
|
||||
auto *next = found == entries.end() ? nullptr : found->second.reporter.get();
|
||||
const bool useSingle = found == entries.end() && single &&
|
||||
(entries.empty() || singleIdentity.empty() || singleIdentity == identity);
|
||||
auto *next = found == entries.end() ? (useSingle ? single.get() : nullptr) : found->second.reporter.get();
|
||||
if (active != next && active) active->finish();
|
||||
active = nullptr; // A failed or unregistered identity must never reuse the old report.
|
||||
if (found == entries.end()) return;
|
||||
if (found == entries.end() && !useSingle) return;
|
||||
try {
|
||||
auto &entry = found->second;
|
||||
if (!entry.reporter) entry.reporter = std::make_unique<ProgressReporter>(entry.path);
|
||||
else entry.reporter->resume();
|
||||
active = entry.reporter.get();
|
||||
if (found == entries.end()) {
|
||||
singleIdentity = identity;
|
||||
single->resume();
|
||||
active = single.get();
|
||||
} else {
|
||||
auto &entry = found->second;
|
||||
if (!entry.reporter) entry.reporter = std::make_unique<ProgressReporter>(entry.path);
|
||||
else entry.reporter->resume();
|
||||
active = entry.reporter.get();
|
||||
}
|
||||
active->begin(identity, live);
|
||||
} catch (const std::exception &exception) {
|
||||
try { if (failure.empty()) failure = exception.what(); } catch (...) {}
|
||||
@@ -331,7 +371,10 @@ void ProgressReports::finish() noexcept {
|
||||
std::string ProgressReports::error() const {
|
||||
std::lock_guard lock(mutex);
|
||||
if (!failure.empty()) return failure;
|
||||
if (single) return single->error();
|
||||
if (single) {
|
||||
const auto message = single->error();
|
||||
if (!message.empty()) return message;
|
||||
}
|
||||
for (const auto &[identity, entry] : entries) {
|
||||
if (entry.reporter) {
|
||||
auto message = entry.reporter->error();
|
||||
|
||||
@@ -73,15 +73,19 @@ class ProgressReports {
|
||||
public:
|
||||
explicit ProgressReports(const std::string &path);
|
||||
ProgressReports() = default;
|
||||
void setSingleFile(const std::string &path);
|
||||
void registerFile(const std::string &identity, const std::string &path);
|
||||
void registerFiles(const char *const *identities, const char *const *paths, int count);
|
||||
void begin(const std::string &identity, bool live = false) noexcept;
|
||||
void observe(const PlayerState &state) noexcept { if (active) active->observe(state); }
|
||||
void closeMedia() noexcept { if (active) active->closeMedia(); }
|
||||
void observe(const PlayerState &state) noexcept;
|
||||
void closeMedia() noexcept;
|
||||
void finish() noexcept;
|
||||
std::string error() const;
|
||||
private:
|
||||
struct Entry { std::string path; std::unique_ptr<ProgressReporter> reporter; };
|
||||
std::unique_ptr<ProgressReporter> single;
|
||||
std::string singlePath;
|
||||
std::string singleIdentity;
|
||||
std::map<std::string, Entry> entries;
|
||||
std::set<std::string> paths;
|
||||
ProgressReporter *active = nullptr;
|
||||
|
||||
Reference in New Issue
Block a user