wav_writer.cpp
Rama main · réplica de lectura del repositorio original.
// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
#include "wav_writer.h"
#include "consumption_diagnostic.h"
#include "queue_item_store.h"
#include "upnp/stream_server.h"
#include <algorithm>
#include <cstring>
#include <cstdio>
#include <cerrno>
#include <limits>
#include <stdexcept>
namespace queue_audio {
uint64_t wavLength(uint64_t frames) {
if (frames > (uint64_t(UINT32_MAX) - 36) / 6)
throw std::invalid_argument("WAV exceeds RIFF limit");
return 44 + frames * 6;
}
std::array<char, 44> wavHeader(uint32_t rate, uint64_t frames) {
if (!rate || rate > UINT32_MAX / 6)
throw std::invalid_argument("invalid WAV rate");
const auto length = wavLength(frames);
std::array<char, 44> h{};
std::memcpy(h.data(), "RIFF", 4);
std::memcpy(h.data() + 8, "WAVEfmt ", 8);
std::memcpy(h.data() + 36, "data", 4);
auto put = [&](unsigned at, uint32_t value, unsigned bytes) {
for (unsigned i = 0; i < bytes; ++i)
h[at + i] = char(value >> (8 * i));
};
put(4, length - 8, 4);
put(16, 16, 4);
put(20, 1, 2);
put(22, 2, 2);
put(24, rate, 4);
put(28, rate * 6, 4);
put(32, 6, 2);
put(34, 24, 2);
put(40, length - 44, 4);
return h;
}
ByteRange wavRange(const std::string &text, uint64_t length) {
if (!length)
throw std::invalid_argument("empty entity");
if (text.empty())
return {0, length - 1, false};
if (text.compare(0, 6, "bytes=") || text.find(',') != std::string::npos)
throw std::invalid_argument("unsupported Range");
const auto dash = text.find('-', 6);
if (dash == std::string::npos)
throw std::invalid_argument("invalid Range");
auto number = [](const std::string &s) -> uint64_t {
if (s.empty() || s.find_first_not_of("0123456789") != std::string::npos)
throw std::invalid_argument("invalid Range number");
return std::stoull(s);
};
const auto a = text.substr(6, dash - 6), b = text.substr(dash + 1);
if (a.empty()) {
auto n = number(b);
if (!n)
throw std::invalid_argument("empty suffix");
return {n >= length ? 0 : length - n, length - 1, true};
}
auto first = number(a), last = b.empty() ? length - 1 : std::min(number(b), length - 1);
if (first >= length || first > last)
throw std::invalid_argument("unsatisfiable Range");
return {first, last, true};
}
bool serveWav(upnp::StreamRequest &request, const std::shared_ptr<Item> &published,
std::chrono::milliseconds wait,
std::function<std::shared_ptr<Item>(upnp::StreamRequest &)> resolve) {
auto item = published;
struct Diagnostic {
upnp::StreamRequest &request;
std::shared_ptr<Item> item;
unsigned status = 0;
uint64_t length = 0, served = 0, bodyStart = UINT64_MAX, backpressure = 0;
std::string ending = "cancelled";
~Diagnostic() {
const auto sent = request.connectionSentBytes();
if (bodyStart != UINT64_MAX && sent != UINT64_MAX && sent >= bodyStart) served = sent-bodyStart;
// Only a valid numeric Range is logged. Never include URLs or tokens.
auto range = request.header("Range");
if (range.find_first_not_of("bytes=0123456789-, ") != std::string::npos || range.size() > 128)
range = "invalid";
std::printf("yeney: queue HTTP method=%s item=%s range=%s status=%u content_length=%llu produced_bytes=%llu served_bytes=%llu end=%s backpressure_retries=%llu\n",
request.method() == upnp::StreamRequest::Method::Get ? "GET" :
request.method() == upnp::StreamRequest::Method::Head ? "HEAD" : "OTHER",
item->descriptor().occurrence.c_str(), range.empty() ? "none" : range.c_str(), status,
(unsigned long long)length, (unsigned long long)(44 + item->producedFrames()*6),
(unsigned long long)served, ending.c_str(), (unsigned long long)backpressure);
}
} diagnostic{request, item};
if (request.method() != upnp::StreamRequest::Method::Get &&
request.method() != upnp::StreamRequest::Method::Head) {
diagnostic.status = 405; diagnostic.ending = "complete";
request.reply(405);
return true;
}
auto length = item->contentLength();
if (!length) {
diagnostic.status = 503; diagnostic.ending = "complete";
request.reply(503);
return true;
} // sealed assets are unpublished
ByteRange range;
try {
range = wavRange(request.header("Range"), *length);
} catch (const std::exception &) {
diagnostic.status = 416; diagnostic.ending = "complete";
const auto wire = "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */" +
std::to_string(*length) + "\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
request.send(wire.data(), wire.size());
return true;
}
diagnostic.status = range.partial ? 206 : 200;
diagnostic.length = range.last - range.first + 1;
auto wire =
std::string("HTTP/1.1 ") + (range.partial ? "206 Partial Content" : "200 OK") +
"\r\nContent-Type: audio/wav\r\nContent-Length: " + std::to_string(range.last - range.first + 1) +
"\r\nAccept-Ranges: bytes\r\nConnection: close\r\n";
if (range.partial)
wire += "Content-Range: bytes " + std::to_string(range.first) + "-" + std::to_string(range.last) +
"/" + std::to_string(*length) + "\r\n";
wire += "\r\n";
auto send = [&](const char *bytes, size_t size) {
while (size && !request.aborted() && !request.peerClosed()) {
const auto before = request.connectionSentBytes();
if (request.send(bytes,size)) return true;
const auto error = errno;
const auto after = request.connectionSentBytes();
if ((error != EAGAIN && error != EWOULDBLOCK && error != ETIMEDOUT) ||
before == UINT64_MAX || after == UINT64_MAX || after < before || after-before > size) {
errno = error;
return false;
}
// A bounded send deadline is a backpressure observation, not EOF.
// The exact socket count fences a partially accepted chunk so no
// bytes are repeated when the healthy speaker starts reading again.
const auto accepted = size_t(after-before);
bytes += accepted;size -= accepted;
(void)item->contentLength(); // intentional flush/eviction cancels the wait
if (!diagnostic.backpressure++)
std::printf("yeney: queue HTTP backpressure item=%s waiting=receiver count_bytes=%llu\n",
item->descriptor().occurrence.c_str(),(unsigned long long)after);
}
return !size;
};
if (!send(wire.data(), wire.size())) {
diagnostic.ending = errno == EAGAIN || errno == EWOULDBLOCK || errno == ETIMEDOUT ? "timeout" : "speaker-closed";
return true;
}
diagnostic.bodyStart = request.connectionSentBytes();
if (request.method() == upnp::StreamRequest::Method::Head) {
diagnostic.ending = "complete";
return true;
}
if (resolve && range.last >= 44) {
// The immutable declaration has already reached the speaker. HEAD and
// rejected ranges never select LMS; an abandoned GET may end here with
// headers only. The room resolver owns settle/selection, never this HTTP
// worker. It must observe peer close and its bounded startup deadline.
std::shared_ptr<Item> produced;
try {
produced = resolve(request);
if (produced && (produced->room() != published->room() ||
produced->descriptor().rate != published->descriptor().rate ||
produced->contentLength() != length ||
produced->descriptor().kind != published->descriptor().kind)) {
diagnostic.ending = "declaration-mismatch";
request.disconnect();
return true;
}
} catch (const std::exception &) {
diagnostic.ending = "resolution-failed";
request.disconnect();
return true;
}
if (!produced || request.aborted() || request.peerClosed()) {
diagnostic.ending = "abandoned";
return true;
}
item = std::move(produced);
diagnostic.item = item;
}
bridge::ConsumptionDiagnostic consumption(request, item->room(), "queue");
auto reader = item->reader(range.first);
uint64_t left = range.last - range.first + 1;
auto deadline = std::chrono::steady_clock::now() + wait;
diagnostic.ending = "speaker-closed";
try {
while (left && !request.aborted() && !request.peerClosed()) {
char bytes[8192];
// Poll the owner connection while waiting for future PCM. The
// startup deadline applies to first body data; later data has a
// fresh bounded wait, not a 15 s total track lifetime.
size_t n;
try {
n = reader.read(bytes, std::min<uint64_t>(left, sizeof(bytes)),
std::chrono::milliseconds(100));
} catch (const ReadTimeout &) {
if (std::chrono::steady_clock::now() >= deadline)
throw;
continue;
}
if (!n) { diagnostic.ending = "early-EOF"; break; }
if (!send(bytes, n)) {
diagnostic.ending = errno == EAGAIN || errno == EWOULDBLOCK || errno == ETIMEDOUT ? "timeout" : "speaker-closed";
break;
}
diagnostic.served += n;
left -= n;
deadline = std::chrono::steady_clock::now() + wait;
}
if (!left) diagnostic.ending = "complete";
else if (request.aborted()) diagnostic.ending = "cancelled";
} catch (const std::exception &e) {
diagnostic.ending = dynamic_cast<const ReadTimeout *>(&e) ? "timeout" : "cancelled";
std::printf("yeney: queue HTTP read closed item=%s offset=%llu reason=%s\n",
item->descriptor().occurrence.c_str(),
static_cast<unsigned long long>(reader.position()), e.what());
request.disconnect();
}
return true;
}
} // namespace queue_audio