Add the LiveKit session wrapper, verified end-to-end against a real room
stplugin::LiveKitSession wraps livekit::Room for exactly one subscribed slot: connect with the wsUrl/lkToken the API client minted, find the chosen participant's camera (and microphone), and hand decoded frames to callback-shaped handlers the OBS adapter can consume directly. Two architectural decisions worth recording, both forced by reading the SDK rather than guessed: 1. Frames come from VideoStream/AudioStream::fromTrack with our own reader threads, NOT from Room::setOnVideoFrameCallback. The dispatcher API is keyed by (participant identity, track NAME), which we cannot know before the track is published -- and disassembling liblivekit.so confirms that both Room::setOnVideoFrameCallback and the dispatcher's own setOnVideoFrameCallback merely record the registration: neither starts a reader for a track that is already subscribed. Registering after the subscription event, which is the only time the track name exists, would therefore have silently produced no video. Taking the shared_ptr<Track> straight off the TrackSubscribedEvent sidesteps the name entirely, and lets us pick the camera by TrackSource (streamer-tools publishes cameras as Source.Camera and screenshares separately -- apps/web/src/avatar/ publish.ts), which is what we actually mean. 2. Every stream operation runs on one owned worker thread, never on a room event thread. The SDK documents that Room::disconnect() from inside a delegate callback deadlocks, and Room's own event dispatch holds a mutex, so delegate callbacks only ever enqueue a command here. VideoStream::Options::capacity is set (3 frames) so the SDK's queue is a drop-oldest ring buffer: a stalled consumer can only fall three frames behind, and what it then sees is the newest frame rather than a backlog. That is the structural answer to the stale-media bug that motivated this plugin. The pure decision-making -- the state machine, track selection, frame geometry validation -- lives in session_types.h/.cpp with no LiveKit or OBS types, so it is unit-testable headlessly (81 checks in test_session, including the publisher-swap and reconnect transitions, plus the real connect() failure paths against the real SDK: unreachable host, garbage token, incomplete config, and destruction mid-connect). test_integration_livekit is the test that proves media actually flows. It publishes a synthetic camera and microphone into a real LiveKit room using the same SDK, subscribes through LiveKitSession, and asserts on the exact fields the OBS adapter will dereference. It skips (exit 0) unless STPLUGIN_IT_* is set, so the three build runners stay green; scripts/livekit-dev-room.py mints the tokens for a local `livekit-server --dev`. Verified locally against livekit-server 1.13.6 in dev mode: integration_livekit: 36 video frames, 323 audio frames, 10 state changes integration_livekit: 32 checks passed covering: connect; subscribe to the named participant's camera; 320x240 I420 frames with three planes, non-null plane pointers and strides >= the frame's own width; 48kHz audio; unpublish -> hasVideo() false, state stays Connected (a dark camera is the placeholder state, never an error) and NO further frames arrive from the dead publisher; republish -> video resumes; clean disconnect. One real finding from that run, now handled: WebRTC ramps a new subscription up from a downscaled spatial layer, so the first frames after (re)subscribing legitimately arrive smaller than what is being published. The OBS adapter must cope with a mid-stream resolution change; the test asserts per-frame geometry rather than the publisher's, and separately asserts the stream does reach full size. Full suite: ctest -> 6/6 passed. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RL8abRmgFXkVASHkkqiJbE
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,370 @@
|
||||
/*
|
||||
streamer-tools OBS Camera Plugin - LiveKit end-to-end integration test
|
||||
Copyright (C) 2026 CyberCoveLLC <jknapp85@gmail.com>
|
||||
|
||||
This program is free software; you can redistribute it and/or modify
|
||||
it under the terms of the GNU General Public License as published by
|
||||
the Free Software Foundation; either version 2 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
This program is distributed in the hope that it will be useful,
|
||||
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
GNU General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU General Public License along
|
||||
with this program. If not, see <https://www.gnu.org/licenses/>
|
||||
*/
|
||||
|
||||
// The one test that proves the session wrapper actually receives media.
|
||||
//
|
||||
// It publishes a synthetic camera into a real LiveKit room using the same
|
||||
// SDK, subscribes to it through stplugin::LiveKitSession, and asserts that
|
||||
// decoded frames arrive with the geometry the OBS adapter is going to hand
|
||||
// to obs_source_output_video. It also drives the publisher-swap sequence
|
||||
// (unpublish, republish) that motivated this whole plugin, and asserts the
|
||||
// wrapper recovers instead of going stale or erroring out.
|
||||
//
|
||||
// It needs a reachable LiveKit server, so it SKIPS (exit 0) unless these are
|
||||
// set -- CI on the three build runners has no server, and this must not turn
|
||||
// into a red build there:
|
||||
//
|
||||
// STPLUGIN_IT_URL ws://127.0.0.1:7880
|
||||
// STPLUGIN_IT_PUBLISH_TOKEN JWT: roomJoin + canPublish for the room
|
||||
// STPLUGIN_IT_SUBSCRIBE_TOKEN JWT: roomJoin + canSubscribe for the room
|
||||
// STPLUGIN_IT_PUBLISHER_IDENTITY the identity in the publish token
|
||||
//
|
||||
// scripts/livekit-dev-room.py mints all four against a `livekit-server --dev`.
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include <livekit/audio_frame.h>
|
||||
#include <livekit/audio_source.h>
|
||||
#include <livekit/livekit.h>
|
||||
#include <livekit/local_audio_track.h>
|
||||
#include <livekit/local_participant.h>
|
||||
#include <livekit/local_track_publication.h>
|
||||
#include <livekit/local_video_track.h>
|
||||
#include <livekit/room.h>
|
||||
#include <livekit/video_frame.h>
|
||||
#include <livekit/video_source.h>
|
||||
|
||||
#include "stplugin/session.h"
|
||||
#include "test_util.h"
|
||||
|
||||
using namespace stplugin;
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr int kWidth = 320;
|
||||
constexpr int kHeight = 240;
|
||||
|
||||
std::string envOrEmpty(const char *name)
|
||||
{
|
||||
const char *value = std::getenv(name);
|
||||
return value ? std::string(value) : std::string();
|
||||
}
|
||||
|
||||
/// A moving horizontal band, so a frozen or stale frame is distinguishable
|
||||
/// from a live one by luma alone.
|
||||
livekit::VideoFrame makeFrame(int tick)
|
||||
{
|
||||
livekit::VideoFrame frame = livekit::VideoFrame::create(kWidth, kHeight, livekit::VideoBufferType::I420);
|
||||
std::uint8_t *data = frame.data();
|
||||
const std::size_t luma = static_cast<std::size_t>(kWidth) * kHeight;
|
||||
std::memset(data, 16, luma);
|
||||
const int band = (tick * 7) % kHeight;
|
||||
std::memset(data + static_cast<std::size_t>(band) * kWidth, 235, kWidth);
|
||||
std::memset(data + luma, 128, frame.dataSize() - luma);
|
||||
return frame;
|
||||
}
|
||||
|
||||
void step(const char *what)
|
||||
{
|
||||
std::printf(" step: %s\n", what);
|
||||
std::fflush(stdout);
|
||||
}
|
||||
|
||||
bool waitFor(const std::function<bool()> &predicate, int timeout_ms)
|
||||
{
|
||||
const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);
|
||||
while (std::chrono::steady_clock::now() < deadline) {
|
||||
if (predicate())
|
||||
return true;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
}
|
||||
return predicate();
|
||||
}
|
||||
|
||||
struct Publisher {
|
||||
livekit::Room room;
|
||||
std::shared_ptr<livekit::VideoSource> video_source;
|
||||
std::shared_ptr<livekit::LocalVideoTrack> video_track;
|
||||
std::shared_ptr<livekit::AudioSource> audio_source;
|
||||
std::shared_ptr<livekit::LocalAudioTrack> audio_track;
|
||||
std::thread pump;
|
||||
std::atomic<bool> stop{false};
|
||||
std::string video_sid;
|
||||
|
||||
bool connect(const std::string &url, const std::string &token)
|
||||
{
|
||||
livekit::RoomOptions options;
|
||||
options.auto_subscribe = false;
|
||||
options.connect_timeout = std::chrono::milliseconds(10000);
|
||||
return room.connect(url, token, options);
|
||||
}
|
||||
|
||||
bool publishVideo()
|
||||
{
|
||||
auto local = room.localParticipant().lock();
|
||||
if (!local)
|
||||
return false;
|
||||
video_source = std::make_shared<livekit::VideoSource>(kWidth, kHeight);
|
||||
video_track = livekit::LocalVideoTrack::createLocalVideoTrack("camera", video_source);
|
||||
livekit::TrackPublishOptions options;
|
||||
options.source = livekit::TrackSource::SOURCE_CAMERA;
|
||||
options.simulcast = false;
|
||||
local->publishTrack(video_track, options);
|
||||
// publishTrack is async server-side; the SID appears on the track once
|
||||
// the publication lands.
|
||||
for (int i = 0; i < 100 && video_track->sid().empty(); ++i)
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
video_sid = video_track->sid();
|
||||
return !video_sid.empty();
|
||||
}
|
||||
|
||||
bool publishAudio()
|
||||
{
|
||||
auto local = room.localParticipant().lock();
|
||||
if (!local)
|
||||
return false;
|
||||
audio_source = std::make_shared<livekit::AudioSource>(48000, 1);
|
||||
audio_track = livekit::LocalAudioTrack::createLocalAudioTrack("microphone", audio_source);
|
||||
livekit::TrackPublishOptions options;
|
||||
options.source = livekit::TrackSource::SOURCE_MICROPHONE;
|
||||
local->publishTrack(audio_track, options);
|
||||
for (int i = 0; i < 100 && audio_track->sid().empty(); ++i)
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
return !audio_track->sid().empty();
|
||||
}
|
||||
|
||||
bool unpublishVideo()
|
||||
{
|
||||
auto local = room.localParticipant().lock();
|
||||
if (!local)
|
||||
return false;
|
||||
// The FFI keys local publications by the *publication* SID, which is
|
||||
// not necessarily the track SID -- unpublishing by track SID throws
|
||||
// "track not found".
|
||||
std::string sid = video_sid;
|
||||
if (video_track && video_track->publication())
|
||||
sid = video_track->publication()->sid();
|
||||
if (sid.empty())
|
||||
return false;
|
||||
try {
|
||||
local->unpublishTrack(sid);
|
||||
} catch (const std::exception &e) {
|
||||
std::printf(" unpublishTrack(%s) threw: %s\n", sid.c_str(), e.what());
|
||||
std::fflush(stdout);
|
||||
return false;
|
||||
}
|
||||
video_sid.clear();
|
||||
video_track.reset();
|
||||
video_source.reset();
|
||||
return true;
|
||||
}
|
||||
|
||||
void startPump()
|
||||
{
|
||||
stop.store(false);
|
||||
pump = std::thread([this] {
|
||||
int tick = 0;
|
||||
std::vector<std::int16_t> pcm(480, 0); // 10ms of 48kHz mono
|
||||
while (!stop.load()) {
|
||||
if (video_source) {
|
||||
livekit::VideoFrame frame = makeFrame(tick);
|
||||
video_source->captureFrame(frame, static_cast<std::int64_t>(tick) * 33333);
|
||||
}
|
||||
if (audio_source) {
|
||||
for (std::size_t i = 0; i < pcm.size(); ++i)
|
||||
pcm[i] = static_cast<std::int16_t>(((tick * 480 + static_cast<int>(i)) % 100) * 100);
|
||||
livekit::AudioFrame audio(pcm, 48000, 1, static_cast<int>(pcm.size()));
|
||||
audio_source->captureFrame(audio);
|
||||
}
|
||||
++tick;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(33));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void stopPump()
|
||||
{
|
||||
stop.store(true);
|
||||
if (pump.joinable())
|
||||
pump.join();
|
||||
}
|
||||
|
||||
~Publisher()
|
||||
{
|
||||
stopPump();
|
||||
room.disconnect(livekit::DisconnectReason::ClientInitiated);
|
||||
}
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
int main()
|
||||
{
|
||||
const std::string url = envOrEmpty("STPLUGIN_IT_URL");
|
||||
const std::string publish_token = envOrEmpty("STPLUGIN_IT_PUBLISH_TOKEN");
|
||||
const std::string subscribe_token = envOrEmpty("STPLUGIN_IT_SUBSCRIBE_TOKEN");
|
||||
const std::string publisher_identity = envOrEmpty("STPLUGIN_IT_PUBLISHER_IDENTITY");
|
||||
|
||||
if (url.empty() || publish_token.empty() || subscribe_token.empty() || publisher_identity.empty()) {
|
||||
std::printf("integration_livekit: SKIPPED (STPLUGIN_IT_* not set)\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
LiveKitSession::globalInitialize();
|
||||
|
||||
auto publisher_holder = std::make_unique<Publisher>();
|
||||
Publisher &publisher = *publisher_holder;
|
||||
step("publisher connect");
|
||||
ST_ASSERT(publisher.connect(url, publish_token));
|
||||
step("publish video");
|
||||
ST_ASSERT(publisher.publishVideo());
|
||||
step("publish audio");
|
||||
ST_ASSERT(publisher.publishAudio());
|
||||
publisher.startPump();
|
||||
|
||||
// --- subscribe through the wrapper under test --------------------------
|
||||
|
||||
std::atomic<int> video_frames{0};
|
||||
std::atomic<int> audio_frames{0};
|
||||
std::atomic<int> bad_frames{0};
|
||||
std::atomic<int> full_size_frames{0};
|
||||
std::atomic<int> last_width{0};
|
||||
std::atomic<int> last_height{0};
|
||||
std::atomic<int> last_planes{0};
|
||||
std::atomic<long long> last_timestamp{0};
|
||||
std::atomic<int> last_sample_rate{0};
|
||||
std::atomic<int> last_channels{0};
|
||||
|
||||
LiveKitSession session;
|
||||
session.setVideoHandler([&](const VideoFrameData &frame) {
|
||||
// Everything the OBS adapter is about to dereference must be sane --
|
||||
// checked against the frame's OWN geometry, not the publisher's.
|
||||
// WebRTC ramps a new subscription up from a downscaled spatial layer,
|
||||
// so the first frames after (re)subscribing legitimately arrive
|
||||
// smaller than what is being published; the adapter has to cope with
|
||||
// a mid-stream resolution change, and so does this assertion.
|
||||
const std::uint32_t chroma_stride = static_cast<std::uint32_t>((frame.width + 1) / 2);
|
||||
const bool sane = frame.width > 0 && frame.height > 0 && frame.format == PixelFormat::I420 &&
|
||||
frame.plane_count == 3 && frame.data != nullptr &&
|
||||
frame.size >= expectedFrameBytes(frame.format, frame.width, frame.height) &&
|
||||
frame.planes[0].data != nullptr && frame.planes[1].data != nullptr &&
|
||||
frame.planes[2].data != nullptr &&
|
||||
frame.planes[0].stride >= static_cast<std::uint32_t>(frame.width) &&
|
||||
frame.planes[1].stride >= chroma_stride && frame.planes[2].stride >= chroma_stride;
|
||||
if (!sane)
|
||||
bad_frames.fetch_add(1);
|
||||
if (frame.width == kWidth && frame.height == kHeight)
|
||||
full_size_frames.fetch_add(1);
|
||||
last_width.store(frame.width);
|
||||
last_height.store(frame.height);
|
||||
last_planes.store(frame.plane_count);
|
||||
last_timestamp.store(static_cast<long long>(frame.timestamp_us));
|
||||
video_frames.fetch_add(1);
|
||||
});
|
||||
session.setAudioHandler([&](const AudioFrameData &frame) {
|
||||
last_sample_rate.store(frame.sample_rate);
|
||||
last_channels.store(frame.channels);
|
||||
audio_frames.fetch_add(1);
|
||||
});
|
||||
|
||||
std::atomic<int> state_changes{0};
|
||||
session.setStateHandler([&](SessionState, const std::string &) { state_changes.fetch_add(1); });
|
||||
|
||||
SessionConfig config;
|
||||
config.ws_url = url;
|
||||
config.token = subscribe_token;
|
||||
config.participant_identity = publisher_identity;
|
||||
config.connect_timeout_ms = 10000;
|
||||
|
||||
step("subscriber connect");
|
||||
ST_ASSERT(session.connect(config));
|
||||
ST_ASSERT(session.state() == SessionState::Connected);
|
||||
|
||||
ST_ASSERT(waitFor([&] { return video_frames.load() >= 15; }, 25000));
|
||||
ST_ASSERT(session.hasVideo());
|
||||
ST_ASSERT(!session.waitingForCamera());
|
||||
ST_ASSERT_EQ(bad_frames.load(), 0);
|
||||
ST_ASSERT_EQ(last_planes.load(), 3);
|
||||
// The stream must actually reach the published resolution, not just
|
||||
// deliver ramp-up frames forever.
|
||||
ST_ASSERT(waitFor([&] { return full_size_frames.load() > 0; }, 20000));
|
||||
ST_ASSERT_EQ(last_width.load(), kWidth);
|
||||
ST_ASSERT_EQ(last_height.load(), kHeight);
|
||||
ST_ASSERT(last_timestamp.load() > 0);
|
||||
ST_ASSERT(session.videoFrameCount() >= 15);
|
||||
|
||||
ST_ASSERT(waitFor([&] { return audio_frames.load() >= 10; }, 20000));
|
||||
ST_ASSERT(session.hasAudio());
|
||||
ST_ASSERT_EQ(last_sample_rate.load(), 48000);
|
||||
ST_ASSERT(last_channels.load() >= 1);
|
||||
|
||||
// --- publisher swap: the bug this plugin exists to make impossible -----
|
||||
|
||||
const int before_swap = video_frames.load();
|
||||
publisher.stopPump();
|
||||
step("unpublish video");
|
||||
ST_ASSERT(publisher.unpublishVideo());
|
||||
|
||||
ST_ASSERT(waitFor([&] { return !session.hasVideo(); }, 15000));
|
||||
// An unpublished camera is the placeholder state, never a failure: the
|
||||
// room connection itself is untouched.
|
||||
ST_ASSERT(session.state() == SessionState::Connected);
|
||||
ST_ASSERT(session.waitingForCamera());
|
||||
|
||||
// No frames may keep arriving from the dead publisher.
|
||||
const int after_unpublish = video_frames.load();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1500));
|
||||
ST_ASSERT_EQ(video_frames.load(), after_unpublish);
|
||||
ST_ASSERT(after_unpublish >= before_swap);
|
||||
|
||||
// Republish, exactly as a reconnecting browser would.
|
||||
step("republish video");
|
||||
ST_ASSERT(publisher.publishVideo());
|
||||
publisher.startPump();
|
||||
|
||||
ST_ASSERT(waitFor([&] { return video_frames.load() >= after_unpublish + 15; }, 25000));
|
||||
ST_ASSERT(session.hasVideo());
|
||||
ST_ASSERT(session.state() == SessionState::Connected);
|
||||
ST_ASSERT_EQ(bad_frames.load(), 0);
|
||||
|
||||
// --- teardown ----------------------------------------------------------
|
||||
|
||||
step("teardown");
|
||||
publisher.stopPump();
|
||||
session.disconnect();
|
||||
ST_ASSERT(session.state() == SessionState::Disconnected);
|
||||
ST_ASSERT(!session.hasVideo());
|
||||
|
||||
// The publisher's Room must be torn down while the SDK is still
|
||||
// initialized, or its FFI disconnect fails on the way out.
|
||||
publisher_holder.reset();
|
||||
LiveKitSession::globalShutdown();
|
||||
|
||||
std::printf("integration_livekit: %d video frames, %d audio frames, %d state changes\n", video_frames.load(),
|
||||
audio_frames.load(), state_changes.load());
|
||||
return st_test_report("integration_livekit");
|
||||
}
|
||||
Reference in New Issue
Block a user