180 lines
7.4 KiB
C++
180 lines
7.4 KiB
C++
#include <gtest/gtest.h>
|
|
#include <mcp/core/Control_Service.hpp>
|
|
#include <mcp/core/runtime/Gallery_Plots.hpp>
|
|
#include <render_common.hpp>
|
|
#include <web_server/src/media/Gallery_Video_Stream.hpp>
|
|
#include <array>
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <ranges>
|
|
#include <stdexcept>
|
|
#include <string>
|
|
|
|
namespace aethera::web::media {
|
|
namespace {
|
|
|
|
using Clock = std::chrono::steady_clock;
|
|
|
|
void corun_until_or_throw(std::string_view operation,
|
|
const std::function<bool()>& completed) {
|
|
std::string failure;
|
|
Task_Graph coordinator{"gallery.video.test.await"};
|
|
coordinator.add("await", [&] {
|
|
const auto deadline = Clock::now() + std::chrono::seconds{10};
|
|
Task_Graph::corun_until([&] {
|
|
if (completed()) return true;
|
|
if (Clock::now() < deadline) return false;
|
|
failure = std::string{operation} + " timed out";
|
|
return true;
|
|
});
|
|
});
|
|
aethera::detail::run_taskflow(coordinator);
|
|
if (!failure.empty()) throw std::runtime_error(failure);
|
|
}
|
|
|
|
void configure_manual_pixel_delivery(
|
|
const std::shared_ptr<mcp::Control_Service>& control,
|
|
std::string_view id) {
|
|
ASSERT_TRUE(control->write_plot_property(id,
|
|
"frame-analysis", "pacing_mode", "manual").value(
|
|
"success", false));
|
|
ASSERT_NO_THROW(corun_until_or_throw("manual Plot policy", [&] {
|
|
const auto diagnostics = control->plot_diagnostics(id);
|
|
if (!diagnostics.contains("frame_policy")) return false;
|
|
const auto& configuration = diagnostics.at("frame_policy").at(
|
|
"configuration");
|
|
return configuration.at("mode") == "manual" &&
|
|
configuration.at("pixel_delivery_enabled").get<bool>();
|
|
}));
|
|
}
|
|
|
|
void schedule_manual_frame(
|
|
const std::shared_ptr<mcp::Control_Service>& control,
|
|
std::string_view id,
|
|
std::uint64_t correlation_sequence) {
|
|
const auto now = Clock::now();
|
|
control->request_frame(id, Frame_Request{
|
|
.issued_at = now,
|
|
.sequence = correlation_sequence,
|
|
.time_milliseconds =
|
|
std::chrono::duration<double, std::milli>(
|
|
now.time_since_epoch()).count(),
|
|
.width = 720,
|
|
.height = 420,
|
|
.source = Frame_Request_Source::immediate
|
|
});
|
|
}
|
|
|
|
TEST(Gallery_Video_Stream,
|
|
Publication_Feedback_Task_Releases_Each_Plots_Independent_Frame) {
|
|
const auto definitions = gallery_plot_definitions();
|
|
const auto found = std::ranges::find_if(
|
|
definitions, [](const Plot_Definition& definition) {
|
|
return definition.dimension == Plot_Dimension::two_d;
|
|
});
|
|
ASSERT_NE(found, definitions.end());
|
|
const auto second = std::ranges::find_if(
|
|
found + 1, definitions.end(), [](const Plot_Definition& definition) {
|
|
return definition.dimension == Plot_Dimension::two_d;
|
|
});
|
|
ASSERT_NE(second, definitions.end());
|
|
const std::string plot_a{found->id};
|
|
const std::string plot_b{second->id};
|
|
auto control_slot = std::make_shared<std::atomic<
|
|
std::shared_ptr<mcp::Control_Service>>>();
|
|
auto stream = Gallery_Video_Stream::create({
|
|
{plot_a, [control_slot, plot_a](Frame_Publication_Feedback feedback) {
|
|
if (const auto control = control_slot->load())
|
|
control->submit_publication_feedback(
|
|
plot_a, std::move(feedback));
|
|
}},
|
|
{plot_b, [control_slot, plot_b](Frame_Publication_Feedback feedback) {
|
|
if (const auto control = control_slot->load())
|
|
control->submit_publication_feedback(
|
|
plot_b, std::move(feedback));
|
|
}}});
|
|
ASSERT_TRUE(stream);
|
|
auto control = mcp::Control_Service::create(mcp::Gallery_Output{
|
|
.publish = [stream](std::string_view id,
|
|
std::shared_ptr<const Gallery_Frame> frame) {
|
|
return stream->accept_frame(id, std::move(frame));
|
|
}});
|
|
control_slot->store(control);
|
|
configure_manual_pixel_delivery(control, plot_a);
|
|
configure_manual_pixel_delivery(control, plot_b);
|
|
constexpr std::size_t frame_count{5};
|
|
std::array<std::atomic_uint64_t, frame_count> sequences{};
|
|
std::atomic_size_t received{};
|
|
std::atomic<std::shared_ptr<const Encoded_Video_Frame>> outstanding{};
|
|
const auto subscription = stream->subscribe(
|
|
"gallery-video-test",
|
|
[&](Gallery_Stream_Frame frame) {
|
|
if (!frame.video) return false;
|
|
const auto index = received.fetch_add(
|
|
1, std::memory_order_acq_rel);
|
|
if (index < sequences.size())
|
|
sequences[index].store(
|
|
frame.video->sequence, std::memory_order_release);
|
|
outstanding.store(std::move(frame.video),
|
|
std::memory_order_release);
|
|
return true;
|
|
},
|
|
[] { return true; },
|
|
[] { return nlohmann::json::object(); });
|
|
|
|
const std::array<std::string_view, frame_count> schedule{
|
|
plot_a, plot_a, plot_b, plot_a, plot_b};
|
|
stream->request_taskflow_trace(1);
|
|
for (std::size_t index = 0; index < schedule.size(); ++index) {
|
|
const auto published_before = control->plot_diagnostics(schedule[index])
|
|
.at("frame_policy").at("lifecycle")
|
|
.at("published_frame_count").get<std::uint64_t>();
|
|
schedule_manual_frame(control, schedule[index], index + 1U);
|
|
ASSERT_NO_THROW(corun_until_or_throw("encoded Plot frame", [&] {
|
|
return received.load(std::memory_order_acquire) > index;
|
|
}));
|
|
ASSERT_NO_THROW(corun_until_or_throw(
|
|
"frame policy publication feedback", [&] {
|
|
const auto diagnostics = control->plot_diagnostics(schedule[index]);
|
|
return diagnostics.at("frame_lifecycle") == "running" &&
|
|
diagnostics.at("frame_policy").at("lifecycle")
|
|
.at("published_frame_count") > published_before;
|
|
}));
|
|
outstanding.store({}, std::memory_order_release);
|
|
}
|
|
|
|
EXPECT_EQ(received.load(std::memory_order_acquire), frame_count);
|
|
for (std::size_t index = 0; index < frame_count; ++index)
|
|
EXPECT_EQ(sequences[index].load(std::memory_order_acquire),
|
|
index + 1U);
|
|
const auto diagnostics = stream->diagnostics();
|
|
EXPECT_EQ(diagnostics.at("received_frame_count"), frame_count);
|
|
EXPECT_EQ(diagnostics.at("processed_frame_count"), frame_count);
|
|
EXPECT_EQ(diagnostics.at("encoded_frame_count"), frame_count);
|
|
EXPECT_EQ(diagnostics.at("pending_frame_count"), 0U);
|
|
|
|
ASSERT_NO_THROW(corun_until_or_throw("gallery Taskflow trace", [&] {
|
|
return stream->taskflow_trace().value("complete", false);
|
|
}));
|
|
const auto trace = stream->taskflow_trace();
|
|
ASSERT_EQ(trace.at("frames").size(), 1U);
|
|
const auto& graphs = trace.at("frames").front().at("graphs");
|
|
const auto graph = std::ranges::find_if(graphs, [](const auto& value) {
|
|
return value.value("name", std::string{}) == "gallery.video.frame";
|
|
});
|
|
ASSERT_NE(graph, graphs.end());
|
|
const auto& nodes = graph->at("nodes");
|
|
const auto feedback = std::ranges::find_if(nodes, [](const auto& value) {
|
|
return value.value("name", std::string{}) ==
|
|
"frame.policy.publish.feedback";
|
|
});
|
|
ASSERT_NE(feedback, nodes.end());
|
|
EXPECT_EQ(feedback->at("attributes").at("owner"), "frame_policy");
|
|
|
|
stream->unsubscribe(subscription);
|
|
stream->shutdown();
|
|
}
|
|
|
|
}
|
|
}
|