You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
289 lines
17 KiB
289 lines
17 KiB
/* Настоящее ядро/actor, подписанный медиаблок и H.264 decoder: каталог до подписки, эпоха, показ и teardown. */ |
|
#include "../src/streamplayer_window.h" |
|
#include "../src/sound_manager.h" |
|
#include "../transport/stream_bridge.h" |
|
#include <QtTest> |
|
#include <QJsonDocument> |
|
#include <QJsonArray> |
|
#include <QLabel> |
|
#include <QSlider> |
|
#include <atomic> |
|
#include <cmath> |
|
#include <chrono> |
|
|
|
extern "C" { |
|
#include "utun_instance.h" |
|
#include "chat/chat_core.h" |
|
#include "chat/chat_event.h" |
|
#include "stream/stream.h" |
|
#include "stream/stream_member.h" |
|
#include "debug_config.h" |
|
#include "mem.h" |
|
#include "sqlite3.h" |
|
#include "opus_codec.h" |
|
#ifdef HAVE_FFMPEG |
|
#include <libavcodec/avcodec.h> |
|
#endif |
|
} |
|
|
|
static QList<QByteArray> snapshots; |
|
#ifdef TEST_NULL_AUDIO |
|
static std::atomic<ma_device_data_proc> originalAudioCallback{nullptr}; |
|
static std::atomic<uint64_t> nonzeroSamples{0}; |
|
static std::atomic<uint64_t> leftEnergy{0}, rightEnergy{0}; |
|
static std::atomic<int64_t> playbackDelayUs{0}; |
|
extern "C" int64_t __real_audio_stream_frame_time(ma_device*, unsigned int, int*); |
|
extern "C" int64_t __wrap_audio_stream_frame_time(ma_device* device, unsigned int frames, int* valid) { |
|
if (!playbackDelayUs) return __real_audio_stream_frame_time(device, frames, valid); |
|
*valid = 1; |
|
return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count() |
|
+ playbackDelayUs.load(); |
|
} |
|
extern "C" ma_result __real_ma_context_init(const ma_backend*, ma_uint32, const ma_context_config*, ma_context*); |
|
extern "C" ma_result __real_ma_device_init(ma_context*, const ma_device_config*, ma_device*); |
|
extern "C" ma_result __wrap_ma_context_init(const ma_backend*, ma_uint32, const ma_context_config*, ma_context* context) { |
|
ma_backend backend = ma_backend_null; return __real_ma_context_init(&backend, 1, nullptr, context); |
|
} |
|
static void observeAudio(ma_device* device, void* out, const void* in, ma_uint32 frames) { |
|
originalAudioCallback.load()(device, out, in, frames); |
|
const auto* pcm = static_cast<const int16_t*>(out); |
|
uint64_t left = 0, right = 0, nonzero = 0; |
|
for (unsigned i = 0; i < frames; i++) { |
|
left += static_cast<int64_t>(pcm[2 * i]) * pcm[2 * i]; |
|
right += static_cast<int64_t>(pcm[2 * i + 1]) * pcm[2 * i + 1]; |
|
nonzero += pcm[2 * i] != 0; nonzero += pcm[2 * i + 1] != 0; |
|
} |
|
leftEnergy.fetch_add(left); rightEnergy.fetch_add(right); nonzeroSamples.fetch_add(nonzero); |
|
} |
|
extern "C" ma_result __wrap_ma_device_init(ma_context* context, const ma_device_config* config, ma_device* device) { |
|
ma_device_config copy = *config; |
|
if (copy.deviceType == ma_device_type_playback && copy.playback.format == ma_format_s16 && copy.playback.channels == 2) { |
|
originalAudioCallback = copy.dataCallback; copy.dataCallback = observeAudio; |
|
} |
|
return __real_ma_device_init(context, ©, device); |
|
} |
|
#endif |
|
static void sourceEvent(UTUN_INSTANCE*, int type, const uint8_t* data, int size) { |
|
if (type == CHAT_EVT_STREAM_SOURCES) snapshots.push_back(QByteArray(reinterpret_cast<const char*>(data), size)); |
|
} |
|
|
|
class StreamViewTest : public QObject { |
|
Q_OBJECT |
|
UASYNC* ua = nullptr; |
|
UTUN_INSTANCE* inst = nullptr; |
|
STREAM_HANDLE* publisher = nullptr; |
|
uint64_t group = 0, epoch = 0, frameId = 0, mediaPts = 0; |
|
QTimer poll; |
|
size_t allocations = 0; |
|
uint32_t revision = 1; |
|
uint16_t rotation = 90; |
|
|
|
bool receive(uint64_t id, StreamRecord& record, int64_t position = -1) { |
|
QElapsedTimer elapsed; elapsed.start(); |
|
while (elapsed.elapsed() < 1000) { |
|
if (gui_stream_pull(id, record, position)) return true; |
|
QTest::qWait(1); |
|
} |
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop test missing record id=%llu", (unsigned long long)id); return false; |
|
} |
|
|
|
void publish(const QByteArray& video, const QList<QByteArray>& audio = {}, unsigned videoFrames = 1) { |
|
stream_media_block media{}; media.format = {160, 90, rotation, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, revision}; |
|
mediaPts += 100000; |
|
media.first_frame_offset = 0; media.video_count = videoFrames; |
|
for (unsigned i = 0; i < videoFrames; i++) media.video[i] = {++frameId, mediaPts + i * 100000 / videoFrames, |
|
static_cast<uint32_t>(video.size()), 0, static_cast<uint32_t>(video.size()), STREAM_MEDIA_KEY, |
|
reinterpret_cast<const uint8_t*>(video.constData())}; |
|
if (!audio.isEmpty()) { |
|
media.audio_count = audio.size(); |
|
for (unsigned i = 0; i < media.audio_count; i++) media.audio[i] = {mediaPts + i * 20000, |
|
static_cast<uint32_t>(audio[i].size()), 960, reinterpret_cast<const uint8_t*>(audio[i].constData())}; |
|
} |
|
uint8_t wire[7080]; size_t length = 0; |
|
QCOMPARE(stream_media_encode(&media, wire, sizeof(wire), &length), STREAM_OK); |
|
QCOMPARE(stream_publish_block(publisher, wire, length), STREAM_OK); |
|
} |
|
|
|
private slots: |
|
void initTestCase() { |
|
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); |
|
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR); |
|
allocations = u_get_allocated_count(); ua = uasync_create(); QVERIFY(ua); |
|
inst = utun_instance_create_from_str(ua, |
|
"[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n"); |
|
QVERIFY(inst); QCOMPARE(utun_core_start(inst), 0); QCOMPARE(chat_service_start(inst), 0); |
|
chat_core_create_channel_auto(inst, "Stream test"); |
|
sqlite3_stmt* statement = nullptr; |
|
QCOMPARE(sqlite3_prepare_v2(chat_core_get_db(inst), "SELECT channel_id FROM channels LIMIT 1", -1, &statement, nullptr), SQLITE_OK); |
|
QCOMPARE(sqlite3_step(statement), SQLITE_ROW); |
|
group = QString::fromUtf8(reinterpret_cast<const char*>(sqlite3_column_text(statement, 0))).toULongLong(); |
|
sqlite3_finalize(statement); QVERIFY(group); |
|
chat_event_set_handler(inst, sourceEvent); QVERIFY(gui_stream_watch(group)); |
|
QVERIFY(gui_stream_attach(inst)); uasync_mark_running(ua); |
|
connect(&poll, &QTimer::timeout, this, [this] { uasync_poll(ua, 0); }); poll.start(1); |
|
} |
|
|
|
void catalogueWithoutViewer() { |
|
QTRY_COMPARE(snapshots.size(), 1); |
|
QCOMPARE(snapshots.last().mid(9), QByteArray("[]")); QVERIFY(!stream_service_needs_activity(inst)); |
|
QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK); |
|
stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); epoch = info.descriptor.source_epoch; |
|
QTRY_COMPARE(snapshots.size(), 2); |
|
auto sources = QJsonDocument::fromJson(snapshots.last().mid(9)).array(); QCOMPARE(sources.size(), 1); |
|
QCOMPARE(stream_read_be(reinterpret_cast<const uint8_t*>(snapshots.last().constData()), 8), group); |
|
QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); |
|
} |
|
|
|
void staleSelectionAndCancellation() { |
|
QVERIFY(gui_stream_open(100, group, inst->node_id, epoch ^ 1)); |
|
StreamRecord event; |
|
QVERIFY(receive(100, event)); QCOMPARE(event.kind, StreamRecordKind::Closed); |
|
QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED); |
|
QVERIFY(gui_stream_open(101, group, inst->node_id, epoch)); gui_stream_stop(101); QTest::qWait(20); |
|
stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); |
|
} |
|
|
|
void mediaAndPressure() { |
|
QVERIFY(gui_stream_open(102, group, inst->node_id, epoch)); StreamRecord event; |
|
QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Opened); |
|
const QByteArray frame(1500, 'k'); const QByteArray audio("opus"); publish(frame, {audio}); |
|
QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video); QCOMPARE(event.data, frame); |
|
QCOMPARE(event.format.rotation, uint16_t(90)); |
|
QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Audio); QCOMPARE(event.data, audio); |
|
for (int i = 0; i < 270; i++) { publish(frame); uasync_poll(ua, 0); } |
|
unsigned records = 0; |
|
QElapsedTimer held; held.start(); |
|
while (held.elapsed() < 30) { |
|
if (gui_stream_pull(102, event, 100000, 0, 0)) { QVERIFY(event.kind != StreamRecordKind::Gap); records++; } |
|
QTest::qWait(1); |
|
} |
|
QVERIFY(records <= 2); /* Накопленные 27 секунд остаются в RS, а не в очереди frontend. */ |
|
QVERIFY(receive(102, event, 200000)); QCOMPARE(event.kind, StreamRecordKind::Video); |
|
QCOMPARE(event.pts, uint64_t(200000)); QCOMPARE(event.data, frame); |
|
gui_stream_stop(102); QTest::qWait(20); |
|
stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); |
|
} |
|
|
|
void realH264RenderingAndClose() { |
|
#ifdef HAVE_FFMPEG |
|
/* Фикстура протокола выше содержит фиктивный H264; реальный decoder получает отдельную эпоху. */ |
|
QCOMPARE(stream_publish_finish(publisher), STREAM_OK); stream_close(publisher); publisher = nullptr; |
|
QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK); |
|
stream_info fresh{}; QCOMPARE(stream_get_info(publisher, &fresh), STREAM_OK); epoch = fresh.descriptor.source_epoch; |
|
mediaPts = 0; frameId = 0; revision = 1; |
|
const AVCodec* encoder = avcodec_find_encoder(AV_CODEC_ID_H264); |
|
if (!encoder) QSKIP("H264 encoder is unavailable for the decoder fixture"); |
|
AVCodecContext* context = avcodec_alloc_context3(encoder); QVERIFY(context); |
|
context->width = 160; context->height = 90; context->pix_fmt = AV_PIX_FMT_YUV420P; |
|
context->time_base = {1, 10}; context->framerate = {10, 1}; context->max_b_frames = 0; |
|
AVDictionary* options = nullptr; av_dict_set(&options, "preset", "ultrafast", 0); av_dict_set(&options, "tune", "zerolatency", 0); |
|
QCOMPARE(avcodec_open2(context, encoder, &options), 0); av_dict_free(&options); |
|
AVFrame* frame = av_frame_alloc(); QVERIFY(frame); frame->width = 160; frame->height = 90; frame->format = context->pix_fmt; |
|
QCOMPARE(av_frame_get_buffer(frame, 32), 0); |
|
for (int plane = 0; plane < 3; plane++) for (int y = 0; y < (plane ? 45 : 90); y++) |
|
memset(frame->data[plane] + y * frame->linesize[plane], plane ? 128 : 80, plane ? 80 : 160); |
|
frame->pts = 0; QCOMPARE(avcodec_send_frame(context, frame), 0); |
|
AVPacket* packet = av_packet_alloc(); QVERIFY(packet); QCOMPARE(avcodec_receive_packet(context, packet), 0); |
|
QByteArray video(4, '\0'); video.append(reinterpret_cast<const char*>(packet->data), packet->size); |
|
av_packet_free(&packet); av_frame_free(&frame); avcodec_free_context(&context); |
|
std::unique_ptr<opus_codec_encoder_t, decltype(&opus_codec_encoder_destroy)> opus(nullptr, opus_codec_encoder_destroy); |
|
uint64_t sampleFrame = 0; |
|
#ifdef TEST_NULL_AUDIO |
|
QVERIFY(SoundManager::instance()->init()); |
|
opus.reset(opus_codec_encoder_create(48000, 2)); QVERIFY(opus); |
|
QCOMPARE(opus_codec_encoder_application_set(opus.get(), OPUS_CODEC_APP_AUDIO), 0); |
|
QCOMPARE(opus_codec_encoder_bitrate_set(opus.get(), 128000), 0); |
|
nonzeroSamples = 0; leftEnergy = rightEnergy = 0; playbackDelayUs = 200000; |
|
#endif |
|
StreamPlayerWindow window(group, inst->node_id, epoch, "Source"); |
|
QSignalSpy shown(&window, &StreamPlayerWindow::framePresented); window.show(); |
|
auto audioPackets = [&opus, &sampleFrame] { |
|
QList<QByteArray> packets; |
|
if (opus) for (unsigned frame = 0; frame < 5; frame++) { |
|
int16_t pcm[1920]; uint8_t encoded[1275]; |
|
for (unsigned i = 0; i < 960; i++) { |
|
pcm[2 * i] = static_cast<int16_t>(6000 * std::sin((sampleFrame + i) * 0.05)); |
|
pcm[2 * i + 1] = static_cast<int16_t>(1500 * std::sin((sampleFrame + i) * 0.13)); |
|
} |
|
sampleFrame += 960; |
|
int length = opus_codec_encode(opus.get(), pcm, 960, encoded, sizeof(encoded)); |
|
if (length <= 0) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop startup fixture Opus encode failed result=%d", length); |
|
return QList<QByteArray>{}; |
|
} |
|
packets.append(QByteArray(reinterpret_cast<const char*>(encoded), length)); |
|
} |
|
return packets; |
|
}; |
|
#ifdef TEST_NULL_AUDIO |
|
QTest::qWait(20); |
|
for (unsigned batch = 0; batch < 8; batch++) { publish({}, audioPackets(), 0); uasync_poll(ua, 0); } |
|
QTest::qWait(200); |
|
QCOMPARE(shown.count(), 0); |
|
QCOMPARE(nonzeroSamples.load(), uint64_t(0)); |
|
#endif |
|
QTimer sender; |
|
connect(&sender, &QTimer::timeout, this, [this, video, &audioPackets] { |
|
publish(video, audioPackets(), 3); |
|
}); |
|
sender.start(100); |
|
QTRY_VERIFY_WITH_TIMEOUT(shown.count() > 0, 5000); |
|
int beforeSteady = shown.count(); QTest::qWait(1500); |
|
QVERIFY2(shown.count() >= beforeSteady + 30, "30fps video must keep advancing with 200ms physical audio latency"); |
|
int beforeSwitch = shown.count(); revision++; rotation = 270; |
|
QTRY_VERIFY_WITH_TIMEOUT(shown.count() > beforeSwitch + 2, 5000); |
|
stream_info switched{}; QCOMPARE(stream_get_info(publisher, &switched), STREAM_OK); |
|
QCOMPARE(switched.descriptor.source_epoch, epoch); QCOMPARE(switched.owners, 2u); |
|
#ifdef TEST_NULL_AUDIO |
|
QTRY_VERIFY_WITH_TIMEOUT(nonzeroSamples.load() > 0, 5000); |
|
QVERIFY(rightEnergy.load() > 0); |
|
qInfo("Stereo playback energies: left=%llu right=%llu ratio=%.3f", |
|
(unsigned long long)leftEnergy.load(), (unsigned long long)rightEnergy.load(), |
|
static_cast<double>(leftEnergy.load()) / rightEnergy.load()); |
|
QVERIFY2(leftEnergy.load() > rightEnergy.load() * 8 && leftEnergy.load() < rightEnergy.load() * 25, |
|
"Independent stereo channels must reach the playback device without downmix"); |
|
auto* volume = window.findChild<QSlider*>(); QVERIFY(volume); volume->setValue(0); |
|
QTest::qWait(50); uint64_t played = nonzeroSamples.load(); QTest::qWait(100); QCOMPARE(nonzeroSamples.load(), played); |
|
#endif |
|
sender.stop(); window.close(); QTest::qWait(20); |
|
#ifdef TEST_NULL_AUDIO |
|
playbackDelayUs = 0; |
|
#endif |
|
stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); |
|
#else |
|
QSKIP("FFmpeg is disabled"); |
|
#endif |
|
} |
|
|
|
void finishAndDetachPendingOwner() { |
|
QCOMPARE(stream_publish_finish(publisher), STREAM_OK); QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]")); |
|
stream_close(publisher); publisher = nullptr; |
|
QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK); |
|
stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QVERIFY(info.descriptor.source_epoch != epoch); |
|
QVERIFY(gui_stream_open(103, group, inst->node_id, epoch)); StreamRecord event; |
|
QVERIFY(receive(103, event)); QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED); |
|
QVERIFY(gui_stream_open(104, group, inst->node_id, info.descriptor.source_epoch)); |
|
gui_stream_detach(inst); QVERIFY(!gui_stream_ready()); QTest::qWait(20); |
|
QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); |
|
} |
|
|
|
void channelClosureNotifiesViewer() { |
|
QVERIFY(gui_stream_attach(inst)); QVERIFY(gui_stream_watch(group)); |
|
stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); |
|
QVERIFY(gui_stream_open(105, group, inst->node_id, info.descriptor.source_epoch)); |
|
StreamRecord event; QVERIFY(receive(105, event)); QCOMPARE(event.kind, StreamRecordKind::Opened); |
|
QByteArray channel = QByteArray::number(group); chat_core_delete_channel(inst, channel.constData()); |
|
QVERIFY(receive(105, event)); QCOMPARE(event.kind, StreamRecordKind::Closed); QCOMPARE(event.reason, STREAM_CLOSE_GROUP); |
|
QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]")); |
|
QVERIFY(!stream_handle_active(publisher)); QCOMPARE(stream_handle_reason(publisher), STREAM_CLOSE_GROUP); |
|
} |
|
|
|
void cleanupTestCase() { |
|
poll.stop(); gui_stream_detach(inst); stream_close(publisher); uasync_mark_stopped(ua); |
|
SoundManager::instance()->shutdown(); |
|
utun_instance_destroy(inst); uasync_destroy(ua, 0); |
|
QCOMPARE(u_get_allocated_count(), allocations); |
|
} |
|
}; |
|
QTEST_MAIN(StreamViewTest) |
|
#include "test_stream_view.moc"
|
|
|