799 lines
30 KiB
C++
799 lines
30 KiB
C++
#include "quic_test_gateway.h"
|
|
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <cctype>
|
|
#include <cstdint>
|
|
#include <functional>
|
|
#include <iostream>
|
|
#include <limits>
|
|
#include <mutex>
|
|
#include <optional>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
#include "common/media/media_frame.h"
|
|
#include "manager/media_source_hub/include/media_source_hub.h"
|
|
#include "service/quic_edge/include/quic_edge_service.h"
|
|
#include "service/quic_edge/include/quic_transport.h"
|
|
|
|
namespace {
|
|
|
|
using Clock = std::chrono::steady_clock;
|
|
using cmvr::media::Codec;
|
|
using cmvr::media::MediaFrame;
|
|
using cmvr::media::MediaFramePtr;
|
|
using cmvr::media::MediaKind;
|
|
using cmvr::media::MediaSourceHub;
|
|
using cmvr::media::PayloadFormat;
|
|
using cmvr::media::TrackDescriptor;
|
|
using cmvr::media::TrackDescriptorPtr;
|
|
using cmvr::quic_edge::QuicEdgeService;
|
|
using cmvr::test::quic_gateway::GatewayOptions;
|
|
using cmvr::test::quic_gateway::QuicTestGateway;
|
|
|
|
constexpr auto kOverallTimeout = std::chrono::seconds(15);
|
|
constexpr auto kPollInterval = std::chrono::milliseconds(10);
|
|
constexpr std::uint64_t kFnv1a64OffsetBasis = 14695981039346656037ULL;
|
|
constexpr std::uint64_t kFnv1a64Prime = 1099511628211ULL;
|
|
|
|
struct Arguments {
|
|
std::string certificate_file;
|
|
std::string private_key_file;
|
|
};
|
|
|
|
bool parseArguments(int argc, char* argv[], Arguments* arguments)
|
|
{
|
|
if (!arguments) return false;
|
|
for (int index = 1; index < argc; ++index) {
|
|
const std::string option = argv[index];
|
|
if ((option != "--cert" && option != "--key") ||
|
|
index + 1 >= argc) {
|
|
return false;
|
|
}
|
|
const std::string value = argv[++index];
|
|
if (value.empty()) return false;
|
|
if (option == "--cert") {
|
|
arguments->certificate_file = value;
|
|
} else {
|
|
arguments->private_key_file = value;
|
|
}
|
|
}
|
|
return !arguments->certificate_file.empty() &&
|
|
!arguments->private_key_file.empty();
|
|
}
|
|
|
|
std::optional<std::uint64_t> jsonUnsigned(
|
|
const std::string& json, const std::string& key)
|
|
{
|
|
const std::string marker = "\"" + key + "\":";
|
|
std::size_t position = json.find(marker);
|
|
if (position == std::string::npos) return std::nullopt;
|
|
position += marker.size();
|
|
while (position < json.size() &&
|
|
std::isspace(static_cast<unsigned char>(json[position]))) {
|
|
++position;
|
|
}
|
|
if (position == json.size() ||
|
|
!std::isdigit(static_cast<unsigned char>(json[position]))) {
|
|
return std::nullopt;
|
|
}
|
|
|
|
std::uint64_t value = 0U;
|
|
while (position < json.size() &&
|
|
std::isdigit(static_cast<unsigned char>(json[position]))) {
|
|
const std::uint64_t digit =
|
|
static_cast<std::uint64_t>(json[position] - '0');
|
|
if (value >
|
|
(std::numeric_limits<std::uint64_t>::max() - digit) / 10U) {
|
|
return std::nullopt;
|
|
}
|
|
value = value * 10U + digit;
|
|
++position;
|
|
}
|
|
return value;
|
|
}
|
|
|
|
std::optional<std::string> jsonString(
|
|
const std::string& json, const std::string& key)
|
|
{
|
|
const std::string marker = "\"" + key + "\":\"";
|
|
std::size_t position = json.find(marker);
|
|
if (position == std::string::npos) return std::nullopt;
|
|
position += marker.size();
|
|
|
|
std::string value;
|
|
bool escaped = false;
|
|
for (; position < json.size(); ++position) {
|
|
const char current = json[position];
|
|
if (escaped) {
|
|
value.push_back(current);
|
|
escaped = false;
|
|
} else if (current == '\\') {
|
|
escaped = true;
|
|
} else if (current == '"') {
|
|
return value;
|
|
} else {
|
|
value.push_back(current);
|
|
}
|
|
}
|
|
return std::nullopt;
|
|
}
|
|
|
|
bool countAtLeast(const std::string& summary,
|
|
const std::string& key,
|
|
const std::uint64_t minimum)
|
|
{
|
|
const auto value = jsonUnsigned(summary, key);
|
|
return value && *value >= minimum;
|
|
}
|
|
|
|
bool containsJsonFragment(const std::string& summary,
|
|
const std::string& fragment)
|
|
{
|
|
return summary.find(fragment) != std::string::npos;
|
|
}
|
|
|
|
template <typename Predicate>
|
|
bool waitUntil(const Clock::time_point deadline, Predicate&& predicate)
|
|
{
|
|
while (Clock::now() < deadline) {
|
|
if (predicate()) return true;
|
|
std::this_thread::sleep_for(kPollInterval);
|
|
}
|
|
return predicate();
|
|
}
|
|
|
|
std::uint64_t monotonicNanoseconds()
|
|
{
|
|
return static_cast<std::uint64_t>(
|
|
std::chrono::duration_cast<std::chrono::nanoseconds>(
|
|
Clock::now().time_since_epoch())
|
|
.count());
|
|
}
|
|
|
|
std::uint64_t fnv1a64(const std::uint8_t* data, const std::size_t size)
|
|
{
|
|
std::uint64_t hash = kFnv1a64OffsetBasis;
|
|
for (std::size_t index = 0; index < size; ++index) {
|
|
hash ^= data[index];
|
|
hash *= kFnv1a64Prime;
|
|
}
|
|
return hash;
|
|
}
|
|
|
|
class SyntheticSource {
|
|
public:
|
|
explicit SyntheticSource(TrackDescriptorPtr descriptor)
|
|
: descriptor_(std::move(descriptor))
|
|
{
|
|
}
|
|
|
|
MediaSourceHub::SourceCallbacks callbacks(const bool video)
|
|
{
|
|
MediaSourceHub::SourceCallbacks callbacks;
|
|
callbacks.start = [this](
|
|
const MediaSourceHub::FrameSink& sink,
|
|
const MediaSourceHub::CancelPredicate& cancelled) {
|
|
if (!sink || (cancelled && cancelled())) return false;
|
|
std::lock_guard lock(mutex_);
|
|
sink_ = sink;
|
|
running_ = true;
|
|
return true;
|
|
};
|
|
callbacks.stop = [this]() {
|
|
std::lock_guard lock(mutex_);
|
|
running_ = false;
|
|
sink_ = {};
|
|
};
|
|
if (video) {
|
|
callbacks.request_key_frame = [this]() {
|
|
std::lock_guard lock(mutex_);
|
|
++key_frame_requests_;
|
|
return running_;
|
|
};
|
|
}
|
|
return callbacks;
|
|
}
|
|
|
|
bool publish(std::vector<std::uint8_t> payload,
|
|
const std::uint64_t sequence,
|
|
const bool key_frame,
|
|
const bool discontinuity = false)
|
|
{
|
|
MediaSourceHub::FrameSink sink;
|
|
{
|
|
std::lock_guard lock(mutex_);
|
|
if (!running_ || !sink_) return false;
|
|
sink = sink_;
|
|
}
|
|
|
|
MediaFrame::Config frame;
|
|
frame.descriptor = descriptor_;
|
|
frame.payload = std::move(payload);
|
|
frame.sequence = sequence;
|
|
frame.source_frame_number = sequence;
|
|
frame.pts = static_cast<std::int64_t>(sequence);
|
|
frame.dts = frame.pts;
|
|
frame.duration = 1;
|
|
frame.capture_time_ns = monotonicNanoseconds();
|
|
frame.key_frame = key_frame;
|
|
frame.discontinuity = discontinuity;
|
|
sink(cmvr::media::makeMediaFrame(std::move(frame)));
|
|
return true;
|
|
}
|
|
|
|
std::uint64_t keyFrameRequests() const
|
|
{
|
|
std::lock_guard lock(mutex_);
|
|
return key_frame_requests_;
|
|
}
|
|
|
|
private:
|
|
TrackDescriptorPtr descriptor_;
|
|
mutable std::mutex mutex_;
|
|
MediaSourceHub::FrameSink sink_;
|
|
bool running_{false};
|
|
std::uint64_t key_frame_requests_{0U};
|
|
};
|
|
|
|
TrackDescriptorPtr makeVideoDescriptor()
|
|
{
|
|
TrackDescriptor::Config descriptor;
|
|
descriptor.id = "synthetic-camera/video/color";
|
|
descriptor.source_id = "synthetic-camera";
|
|
descriptor.kind = MediaKind::VIDEO;
|
|
descriptor.codec = Codec::H264;
|
|
descriptor.payload_format = PayloadFormat::ANNEX_B;
|
|
descriptor.time_base = {1, 90000};
|
|
descriptor.width = 640U;
|
|
descriptor.height = 360U;
|
|
descriptor.nominal_rate = 30U;
|
|
descriptor.generation = 0x0000000100000001ULL;
|
|
descriptor.codec_config = {
|
|
0x00U, 0x00U, 0x00U, 0x01U, 0x67U, 0x42U, 0x00U, 0x1EU};
|
|
return cmvr::media::makeTrackDescriptor(std::move(descriptor));
|
|
}
|
|
|
|
TrackDescriptorPtr makeAudioDescriptor()
|
|
{
|
|
TrackDescriptor::Config descriptor;
|
|
descriptor.id = "synthetic-microphone/audio/main";
|
|
descriptor.source_id = "synthetic-microphone";
|
|
descriptor.kind = MediaKind::AUDIO;
|
|
descriptor.codec = Codec::AAC;
|
|
descriptor.payload_format = PayloadFormat::AAC_ADTS;
|
|
descriptor.time_base = {1, 48000};
|
|
descriptor.sample_rate = 48000U;
|
|
descriptor.channels = 2U;
|
|
descriptor.nominal_rate = 50U;
|
|
descriptor.generation = 0x0000000100000002ULL;
|
|
descriptor.codec_config = {0x11U, 0x90U};
|
|
return cmvr::media::makeTrackDescriptor(std::move(descriptor));
|
|
}
|
|
|
|
std::vector<std::uint8_t> h264Payload(const std::uint8_t fill)
|
|
{
|
|
std::vector<std::uint8_t> payload(2500U, fill);
|
|
payload[0] = 0x00U;
|
|
payload[1] = 0x00U;
|
|
payload[2] = 0x00U;
|
|
payload[3] = 0x01U;
|
|
payload[4] = 0x65U; // IDR slice.
|
|
return payload;
|
|
}
|
|
|
|
std::vector<std::uint8_t> aacPayload(const std::uint8_t fill)
|
|
{
|
|
std::vector<std::uint8_t> payload(512U, fill);
|
|
// AAC-LC, 48 kHz, stereo ADTS header for a synthetic access unit.
|
|
const std::uint16_t frame_length =
|
|
static_cast<std::uint16_t>(payload.size());
|
|
payload[0] = 0xFFU;
|
|
payload[1] = 0xF1U;
|
|
payload[2] = 0x4CU;
|
|
payload[3] = static_cast<std::uint8_t>(
|
|
0x80U | ((frame_length >> 11U) & 0x03U));
|
|
payload[4] = static_cast<std::uint8_t>((frame_length >> 3U) & 0xFFU);
|
|
payload[5] = static_cast<std::uint8_t>(
|
|
((frame_length & 0x07U) << 5U) | 0x1FU);
|
|
payload[6] = 0xFCU;
|
|
return payload;
|
|
}
|
|
|
|
cmvr::config::QuicEdgeConfig makeEdgeConfig(const std::uint16_t port)
|
|
{
|
|
cmvr::config::QuicEdgeConfig config;
|
|
config.set_id("quic-real-e2e");
|
|
config.set_server_host("127.0.0.1");
|
|
config.set_server_port(port);
|
|
config.set_alpn("cmvr-quic-edge/1");
|
|
config.set_node_id("cmvr-real-e2e-node");
|
|
config.set_software_version("e2e");
|
|
config.set_grpc_endpoint_host("127.0.0.1");
|
|
config.set_grpc_endpoint_port(50052U);
|
|
config.set_grpc_endpoint_tls(false);
|
|
config.set_include_loopback_interfaces(true);
|
|
config.set_heartbeat_interval_ms(250U);
|
|
config.set_control_response_timeout_ms(1500U);
|
|
config.mutable_tls()->set_allow_insecure(true);
|
|
config.mutable_reconnect()->set_initial_delay_ms(50U);
|
|
config.mutable_reconnect()->set_maximum_delay_ms(500U);
|
|
config.mutable_reconnect()->set_multiplier(2.0);
|
|
config.mutable_reconnect()->set_jitter_percent(0U);
|
|
config.mutable_reconnect()->set_connect_timeout_ms(2000U);
|
|
config.set_maximum_datagram_bytes(1200U);
|
|
config.set_maximum_control_frame_bytes(1024U * 1024U);
|
|
config.set_maximum_frame_bytes(16U * 1024U);
|
|
config.set_datagram_send_queue_depth(128U);
|
|
config.set_media_poll_interval_ms(1U);
|
|
|
|
auto* video = config.add_tracks();
|
|
video->set_track_id(1U);
|
|
video->set_source_kind(
|
|
cmvr::config::QuicEdgeTrackConfig::SOURCE_KIND_CAMERA);
|
|
video->set_device_id("synthetic-camera");
|
|
video->set_source_track_id("synthetic-camera/video/color");
|
|
video->set_enable(true);
|
|
|
|
auto* audio = config.add_tracks();
|
|
audio->set_track_id(2U);
|
|
audio->set_source_kind(
|
|
cmvr::config::QuicEdgeTrackConfig::SOURCE_KIND_MICROPHONE);
|
|
audio->set_device_id("synthetic-microphone");
|
|
audio->set_source_track_id("synthetic-microphone/audio/main");
|
|
audio->set_enable(true);
|
|
return config;
|
|
}
|
|
|
|
cmvr::device::DeviceManagerSnapshot makeDeviceManagerSnapshot()
|
|
{
|
|
cmvr::device::DeviceManagerSnapshot snapshot;
|
|
snapshot.name = "cmvr-real-e2e-manager";
|
|
snapshot.version = "e2e";
|
|
snapshot.description = "synthetic DeviceManager heartbeat snapshot";
|
|
|
|
cmvr::device::ManagedDeviceSnapshot video;
|
|
video.id = "synthetic-camera";
|
|
video.kind = cmvr::device::DeviceKind::Camera;
|
|
video.type_name = "SyntheticCamera";
|
|
video.enabled = true;
|
|
video.state = cmvr::device::ManagedDeviceState::Running;
|
|
video.health.state = cmvr::device::DeviceHealthState::Healthy;
|
|
video.status_updated_at_unix_ms = 1001U;
|
|
snapshot.devices.push_back(video);
|
|
|
|
cmvr::device::ManagedDeviceSnapshot audio;
|
|
audio.id = "synthetic-microphone";
|
|
audio.kind = cmvr::device::DeviceKind::Microphone;
|
|
audio.type_name = "SyntheticMicrophone";
|
|
audio.enabled = true;
|
|
audio.state = cmvr::device::ManagedDeviceState::Error;
|
|
audio.health.state = cmvr::device::DeviceHealthState::Fault;
|
|
audio.abnormal = true;
|
|
audio.error_message = "synthetic health fault";
|
|
audio.status_updated_at_unix_ms = 1002U;
|
|
snapshot.devices.push_back(audio);
|
|
|
|
cmvr::device::ManagedDeviceSnapshot disabled;
|
|
disabled.id = "synthetic-disabled-arm";
|
|
disabled.kind = cmvr::device::DeviceKind::Arm;
|
|
disabled.type_name = "DEVICE_TYPE_ROBOT_ARM";
|
|
disabled.enabled = false;
|
|
disabled.state = cmvr::device::ManagedDeviceState::Disabled;
|
|
disabled.health.state = cmvr::device::DeviceHealthState::Unknown;
|
|
disabled.status_updated_at_unix_ms = 1003U;
|
|
snapshot.devices.push_back(disabled);
|
|
return snapshot;
|
|
}
|
|
|
|
class GatewayStopGuard {
|
|
public:
|
|
explicit GatewayStopGuard(QuicTestGateway* gateway) : gateway_(gateway) {}
|
|
~GatewayStopGuard()
|
|
{
|
|
if (gateway_) gateway_->stop();
|
|
}
|
|
|
|
private:
|
|
QuicTestGateway* gateway_;
|
|
};
|
|
|
|
class ServiceStopGuard {
|
|
public:
|
|
explicit ServiceStopGuard(QuicEdgeService* service) : service_(service) {}
|
|
~ServiceStopGuard()
|
|
{
|
|
if (service_) service_->stop();
|
|
}
|
|
|
|
private:
|
|
QuicEdgeService* service_;
|
|
};
|
|
|
|
int run(const Arguments& arguments)
|
|
{
|
|
GatewayOptions gateway_options;
|
|
gateway_options.bind_address = "127.0.0.1";
|
|
gateway_options.port = 0U;
|
|
gateway_options.certificate_file = arguments.certificate_file;
|
|
gateway_options.private_key_file = arguments.private_key_file;
|
|
gateway_options.heartbeat_interval_ms = 250U;
|
|
gateway_options.maximum_reassembly_bytes = 4U * 1024U * 1024U;
|
|
gateway_options.maximum_reassembly_frames = 32U;
|
|
gateway_options.maximum_frame_bytes = 16U * 1024U;
|
|
gateway_options.maximum_work_queue_bytes = 4U * 1024U * 1024U;
|
|
gateway_options.reassembly_timeout_ms = 1000U;
|
|
|
|
QuicTestGateway gateway(std::move(gateway_options));
|
|
std::string error;
|
|
if (!gateway.start(&error)) {
|
|
std::cerr << "gateway start failed: " << error << '\n';
|
|
return 1;
|
|
}
|
|
GatewayStopGuard gateway_guard(&gateway);
|
|
if (gateway.boundPort() == 0U) {
|
|
std::cerr << "gateway did not publish an ephemeral UDP port\n";
|
|
return 1;
|
|
}
|
|
|
|
const TrackDescriptorPtr video_descriptor = makeVideoDescriptor();
|
|
const TrackDescriptorPtr audio_descriptor = makeAudioDescriptor();
|
|
SyntheticSource video(video_descriptor);
|
|
SyntheticSource audio(audio_descriptor);
|
|
MediaSourceHub hub;
|
|
if (!hub.registerSource(
|
|
video_descriptor, video.callbacks(true), 8U) ||
|
|
!hub.registerSource(
|
|
audio_descriptor, audio.callbacks(false), 8U)) {
|
|
std::cerr << "failed to register synthetic MediaSourceHub tracks\n";
|
|
return 1;
|
|
}
|
|
|
|
auto config = makeEdgeConfig(gateway.boundPort());
|
|
std::string validation_error;
|
|
if (!QuicEdgeService::validateConfig(config, &validation_error)) {
|
|
std::cerr << "invalid E2E edge config: " << validation_error << '\n';
|
|
return 1;
|
|
}
|
|
|
|
auto transport = cmvr::quic_edge::createDefaultQuicTransport(
|
|
config.datagram_send_queue_depth());
|
|
const auto device_manager_snapshot = makeDeviceManagerSnapshot();
|
|
QuicEdgeService service(
|
|
config, std::move(transport), hub,
|
|
[device_manager_snapshot]() { return device_manager_snapshot; });
|
|
ServiceStopGuard service_guard(&service);
|
|
if (!service.initialize(&error)) {
|
|
std::cerr << "edge initialize failed: " << error << '\n';
|
|
return 1;
|
|
}
|
|
if (!service.start(&error)) {
|
|
std::cerr << "edge start failed: " << error << '\n';
|
|
return 1;
|
|
}
|
|
|
|
const Clock::time_point deadline = Clock::now() + kOverallTimeout;
|
|
const bool sources_ready = waitUntil(deadline, [&]() {
|
|
return gateway.hasRuntimeFailure() ||
|
|
(hub.subscriberCount(video_descriptor->id) == 1U &&
|
|
hub.subscriberCount(audio_descriptor->id) == 1U &&
|
|
service.status().registered);
|
|
});
|
|
if (!sources_ready || gateway.hasRuntimeFailure() ||
|
|
hub.subscriberCount(video_descriptor->id) != 1U ||
|
|
hub.subscriberCount(audio_descriptor->id) != 1U) {
|
|
std::cerr << "registration/source subscription timeout; edge_error="
|
|
<< service.lastError() << " gateway_error="
|
|
<< gateway.lastError() << " summary="
|
|
<< gateway.summaryJson() << '\n';
|
|
return 1;
|
|
}
|
|
|
|
// A first frame causes the edge to announce its track. Its DATAGRAM may
|
|
// legally overtake the reliable descriptor and be discarded by the
|
|
// receiver, so these frames are discovery traffic only.
|
|
if (!video.publish(h264Payload(0x31U), 1U, true, true) ||
|
|
!audio.publish(aacPayload(0x41U), 1U, false, true)) {
|
|
std::cerr << "failed to publish discovery synthetic frames\n";
|
|
return 1;
|
|
}
|
|
|
|
const bool descriptors_ready = waitUntil(deadline, [&]() {
|
|
const std::string summary = gateway.summaryJson();
|
|
return gateway.hasRuntimeFailure() ||
|
|
(countAtLeast(summary, "media_sessions_opened", 1U) &&
|
|
countAtLeast(summary, "track_descriptors_received", 2U) &&
|
|
jsonUnsigned(summary, "video_track_id").value_or(0U) == 1U &&
|
|
jsonUnsigned(summary, "audio_track_id").value_or(0U) == 2U &&
|
|
service.stats().frames_queued >= 2U);
|
|
});
|
|
if (!descriptors_ready || gateway.hasRuntimeFailure()) {
|
|
std::cerr << "media descriptor discovery timeout; edge_error="
|
|
<< service.lastError() << " gateway_error="
|
|
<< gateway.lastError() << " summary="
|
|
<< gateway.summaryJson() << '\n';
|
|
return 1;
|
|
}
|
|
|
|
// Once the descriptors are installed, establish one completed frame per
|
|
// track. Independent baselines and exact sequence observations prevent
|
|
// delayed discovery traffic on one track from masking a missing frame on
|
|
// the other.
|
|
const std::string discovery_summary = gateway.summaryJson();
|
|
const std::uint64_t baseline_video_frames =
|
|
jsonUnsigned(
|
|
discovery_summary, "video_frames_completed").value_or(0U);
|
|
const std::uint64_t baseline_audio_frames =
|
|
jsonUnsigned(
|
|
discovery_summary, "audio_frames_completed").value_or(0U);
|
|
|
|
if (!video.publish(h264Payload(0x32U), 2U, true) ||
|
|
!audio.publish(aacPayload(0x42U), 2U, false)) {
|
|
std::cerr << "failed to publish priming synthetic frames\n";
|
|
return 1;
|
|
}
|
|
|
|
const bool priming_complete = waitUntil(deadline, [&]() {
|
|
const std::string summary = gateway.summaryJson();
|
|
return gateway.hasRuntimeFailure() ||
|
|
(countAtLeast(
|
|
summary, "video_frames_completed",
|
|
baseline_video_frames + 1U) &&
|
|
countAtLeast(
|
|
summary, "audio_frames_completed",
|
|
baseline_audio_frames + 1U) &&
|
|
jsonUnsigned(
|
|
summary, "maximum_video_frame_sequence").value_or(0U) >=
|
|
2U &&
|
|
jsonUnsigned(
|
|
summary, "maximum_audio_frame_sequence").value_or(0U) >=
|
|
2U &&
|
|
service.stats().frames_queued >= 4U);
|
|
});
|
|
if (!priming_complete || gateway.hasRuntimeFailure()) {
|
|
std::cerr << "media priming timeout; edge_error="
|
|
<< service.lastError() << " gateway_error="
|
|
<< gateway.lastError() << " summary="
|
|
<< gateway.summaryJson() << '\n';
|
|
return 1;
|
|
}
|
|
|
|
const std::string primed_summary = gateway.summaryJson();
|
|
const std::uint64_t primed_video_frames =
|
|
jsonUnsigned(
|
|
primed_summary, "video_frames_completed").value_or(0U);
|
|
const std::uint64_t primed_audio_frames =
|
|
jsonUnsigned(
|
|
primed_summary, "audio_frames_completed").value_or(0U);
|
|
const std::vector<std::uint8_t> expected_video = h264Payload(0x33U);
|
|
const std::vector<std::uint8_t> expected_audio = aacPayload(0x43U);
|
|
|
|
if (!video.publish(expected_video, 3U, true) ||
|
|
!audio.publish(expected_audio, 3U, false)) {
|
|
std::cerr << "failed to publish validation synthetic frames\n";
|
|
return 1;
|
|
}
|
|
|
|
const bool completed = waitUntil(deadline, [&]() {
|
|
const std::string summary = gateway.summaryJson();
|
|
return gateway.hasRuntimeFailure() ||
|
|
(service.stats().registrations_accepted >= 1U &&
|
|
service.stats().heartbeats_acknowledged >= 2U &&
|
|
service.stats().media_sessions_opened >= 1U &&
|
|
service.stats().frames_queued >= 6U &&
|
|
service.stats().datagrams_queued >= 6U &&
|
|
countAtLeast(summary, "registrations_accepted", 1U) &&
|
|
countAtLeast(summary, "heartbeats_received", 2U) &&
|
|
countAtLeast(summary, "heartbeat_acks_sent", 2U) &&
|
|
countAtLeast(summary, "datagrams_received", 6U) &&
|
|
countAtLeast(
|
|
summary, "video_frames_completed",
|
|
primed_video_frames + 1U) &&
|
|
countAtLeast(
|
|
summary, "audio_frames_completed",
|
|
primed_audio_frames + 1U) &&
|
|
jsonUnsigned(
|
|
summary, "maximum_video_frame_sequence").value_or(0U) ==
|
|
3U &&
|
|
jsonUnsigned(
|
|
summary, "maximum_audio_frame_sequence").value_or(0U) ==
|
|
3U &&
|
|
jsonUnsigned(
|
|
summary, "maximum_video_frame_hash").value_or(0U) ==
|
|
fnv1a64(expected_video.data(), expected_video.size()) &&
|
|
jsonUnsigned(
|
|
summary, "maximum_audio_frame_hash").value_or(0U) ==
|
|
fnv1a64(expected_audio.data(), expected_audio.size()));
|
|
});
|
|
if (!completed || gateway.hasRuntimeFailure()) {
|
|
std::cerr << "real QUIC completion timeout; edge_state="
|
|
<< cmvr::quic_edge::toString(service.state())
|
|
<< " edge_error=" << service.lastError()
|
|
<< " gateway_error=" << gateway.lastError()
|
|
<< " summary=" << gateway.summaryJson() << '\n';
|
|
return 1;
|
|
}
|
|
|
|
const auto edge_status = service.status();
|
|
if (!edge_status.registered || edge_status.session_id.empty() ||
|
|
edge_status.active_media_tracks != 2U ||
|
|
video.keyFrameRequests() == 0U) {
|
|
std::cerr << "edge did not retain the expected session/media state\n";
|
|
return 1;
|
|
}
|
|
|
|
const std::string live_summary = gateway.summaryJson();
|
|
const auto gateway_session =
|
|
jsonString(live_summary, "last_session_id");
|
|
if (!gateway_session || *gateway_session != edge_status.session_id ||
|
|
jsonString(live_summary, "last_grpc_endpoint_host")
|
|
.value_or("") != "127.0.0.1" ||
|
|
jsonUnsigned(live_summary, "last_grpc_endpoint_port")
|
|
.value_or(0U) != 50052U ||
|
|
jsonString(live_summary, "last_observed_source_ip")
|
|
.value_or("") != "127.0.0.1" ||
|
|
!countAtLeast(live_summary, "registration_interface_count", 1U) ||
|
|
!countAtLeast(live_summary, "heartbeat_interface_count", 1U) ||
|
|
jsonString(live_summary, "heartbeat_device_manager_name")
|
|
.value_or("") != "cmvr-real-e2e-manager" ||
|
|
jsonString(live_summary, "heartbeat_device_manager_version")
|
|
.value_or("") != "e2e" ||
|
|
jsonUnsigned(live_summary, "heartbeat_device_count")
|
|
.value_or(0U) != 2U ||
|
|
jsonUnsigned(live_summary, "heartbeat_enabled_device_count")
|
|
.value_or(0U) != 2U ||
|
|
jsonUnsigned(live_summary, "heartbeat_disabled_device_count")
|
|
.value_or(0U) != 0U ||
|
|
jsonUnsigned(live_summary, "heartbeat_error_device_count")
|
|
.value_or(0U) != 1U ||
|
|
jsonUnsigned(live_summary, "heartbeat_unknown_health_device_count")
|
|
.value_or(0U) != 0U ||
|
|
!containsJsonFragment(
|
|
live_summary,
|
|
R"({"device_id":"synthetic-camera","kind":5,"type_name":"SyntheticCamera","enabled":true,"manager_state":5,"health":1,"has_error":false,"error_message":"","status_updated_at_unix_ms":1001})") ||
|
|
containsJsonFragment(
|
|
live_summary,
|
|
R"("device_id":"synthetic-disabled-arm")") ||
|
|
!containsJsonFragment(
|
|
live_summary,
|
|
R"({"device_id":"synthetic-microphone","kind":9,"type_name":"SyntheticMicrophone","enabled":true,"manager_state":7,"health":3,"has_error":true,"error_message":"synthetic health fault","status_updated_at_unix_ms":1002})") ||
|
|
!countAtLeast(live_summary, "track_descriptors_received", 2U) ||
|
|
jsonUnsigned(live_summary, "video_track_id").value_or(0U) != 1U ||
|
|
jsonString(live_summary, "video_device_id").value_or("") !=
|
|
"synthetic-camera" ||
|
|
jsonString(live_summary, "video_source_track_id").value_or("") !=
|
|
video_descriptor->id ||
|
|
jsonString(live_summary, "video_codec").value_or("") != "h264" ||
|
|
jsonString(live_summary, "video_payload_format").value_or("") !=
|
|
"annex_b" ||
|
|
jsonUnsigned(live_summary, "video_codec_generation").value_or(0U) !=
|
|
video_descriptor->generation ||
|
|
jsonUnsigned(
|
|
live_summary, "video_codec_generation_token").value_or(0U) !=
|
|
cmvr::quic_edge::descriptorGenerationToken(
|
|
video_descriptor->generation) ||
|
|
jsonUnsigned(live_summary, "video_width").value_or(0U) != 640U ||
|
|
jsonUnsigned(live_summary, "video_height").value_or(0U) != 360U ||
|
|
jsonUnsigned(
|
|
live_summary, "video_frames_per_second").value_or(0U) != 30U ||
|
|
jsonUnsigned(
|
|
live_summary, "video_codec_config_bytes").value_or(0U) !=
|
|
video_descriptor->codec_config.size() ||
|
|
jsonUnsigned(
|
|
live_summary, "video_codec_config_hash").value_or(0U) !=
|
|
fnv1a64(
|
|
video_descriptor->codec_config.data(),
|
|
video_descriptor->codec_config.size()) ||
|
|
jsonUnsigned(live_summary, "audio_track_id").value_or(0U) != 2U ||
|
|
jsonString(live_summary, "audio_device_id").value_or("") !=
|
|
"synthetic-microphone" ||
|
|
jsonString(live_summary, "audio_source_track_id").value_or("") !=
|
|
audio_descriptor->id ||
|
|
jsonString(live_summary, "audio_codec").value_or("") != "aac" ||
|
|
jsonString(live_summary, "audio_payload_format").value_or("") !=
|
|
"aac_adts" ||
|
|
jsonUnsigned(live_summary, "audio_codec_generation").value_or(0U) !=
|
|
audio_descriptor->generation ||
|
|
jsonUnsigned(
|
|
live_summary, "audio_codec_generation_token").value_or(0U) !=
|
|
cmvr::quic_edge::descriptorGenerationToken(
|
|
audio_descriptor->generation) ||
|
|
jsonUnsigned(live_summary, "audio_sample_rate").value_or(0U) !=
|
|
48000U ||
|
|
jsonUnsigned(live_summary, "audio_channels").value_or(0U) != 2U ||
|
|
jsonUnsigned(
|
|
live_summary, "audio_codec_config_bytes").value_or(0U) !=
|
|
audio_descriptor->codec_config.size() ||
|
|
jsonUnsigned(
|
|
live_summary, "audio_codec_config_hash").value_or(0U) !=
|
|
fnv1a64(
|
|
audio_descriptor->codec_config.data(),
|
|
audio_descriptor->codec_config.size()) ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_video_frame_sequence").value_or(0U) !=
|
|
3U ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_video_frame_track_id").value_or(0U) !=
|
|
1U ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_video_frame_bytes").value_or(0U) !=
|
|
expected_video.size() ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_video_frame_hash").value_or(0U) !=
|
|
fnv1a64(expected_video.data(), expected_video.size()) ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_video_frame_flags").value_or(0U) !=
|
|
cmvr::quic_edge::DATAGRAM_FLAG_KEY_FRAME ||
|
|
jsonUnsigned(
|
|
live_summary,
|
|
"maximum_video_capture_timestamp_us").value_or(0U) == 0U ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_audio_frame_sequence").value_or(0U) !=
|
|
3U ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_audio_frame_track_id").value_or(0U) !=
|
|
2U ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_audio_frame_bytes").value_or(0U) !=
|
|
expected_audio.size() ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_audio_frame_hash").value_or(0U) !=
|
|
fnv1a64(expected_audio.data(), expected_audio.size()) ||
|
|
jsonUnsigned(
|
|
live_summary, "maximum_audio_frame_flags").value_or(1U) !=
|
|
cmvr::quic_edge::DATAGRAM_FLAG_NONE ||
|
|
jsonUnsigned(
|
|
live_summary,
|
|
"maximum_audio_capture_timestamp_us").value_or(0U) == 0U ||
|
|
!countAtLeast(live_summary, "video_frames_completed", 2U) ||
|
|
!countAtLeast(live_summary, "audio_frames_completed", 2U) ||
|
|
!countAtLeast(live_summary, "frame_bytes_completed", 1U) ||
|
|
jsonUnsigned(live_summary, "protocol_violations").value_or(1U) != 0U) {
|
|
std::cerr << "gateway summary validation failed: "
|
|
<< live_summary << '\n';
|
|
return 1;
|
|
}
|
|
|
|
// Close the client first so the gateway can synchronously drain and join
|
|
// its bounded worker without retaining a live connection context.
|
|
service.stop();
|
|
gateway.stop();
|
|
if (gateway.hasRuntimeFailure()) {
|
|
std::cerr << "gateway shutdown failed: " << gateway.lastError() << '\n';
|
|
return 1;
|
|
}
|
|
std::string restart_error;
|
|
if (gateway.start(&restart_error) || restart_error.empty()) {
|
|
std::cerr << "single-use gateway unexpectedly restarted\n";
|
|
return 1;
|
|
}
|
|
|
|
std::cout << "cmvr_quic_msquic_e2e_test: PASS "
|
|
<< gateway.summaryJson() << '\n';
|
|
return 0;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
int main(int argc, char* argv[])
|
|
{
|
|
Arguments arguments;
|
|
if (!parseArguments(argc, argv, &arguments)) {
|
|
std::cerr << "Usage: " << argv[0]
|
|
<< " --cert SERVER_CERT.pem --key SERVER_KEY.pem\n";
|
|
return 2;
|
|
}
|
|
try {
|
|
return run(arguments);
|
|
} catch (const std::exception& error) {
|
|
std::cerr << "unexpected E2E exception: " << error.what() << '\n';
|
|
return 1;
|
|
} catch (...) {
|
|
std::cerr << "unexpected non-standard E2E exception\n";
|
|
return 1;
|
|
}
|
|
}
|