diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt index a8bf2ec..b13f033 100644 --- a/core/CMakeLists.txt +++ b/core/CMakeLists.txt @@ -9,6 +9,8 @@ set(STPLUGIN_CORE_SOURCES src/json.cpp src/http_common.cpp src/api_client.cpp + src/session_types.cpp + src/session.cpp ) # HTTP backend, one per platform. See core/include/stplugin/http.h for why diff --git a/core/include/stplugin/session.h b/core/include/stplugin/session.h new file mode 100644 index 0000000..ae43f35 --- /dev/null +++ b/core/include/stplugin/session.h @@ -0,0 +1,118 @@ +/* +streamer-tools OBS Camera Plugin - LiveKit session wrapper +Copyright (C) 2026 CyberCoveLLC + +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 +*/ + +#pragma once + +#include +#include + +#include "stplugin/session_types.h" + +namespace stplugin { + +struct SessionConfig { + /// LiveKit websocket URL, from POST /api/obs/:slug/token. + std::string ws_url; + /// LiveKit JWT, from the same call. + std::string token; + /// The slot's participant identity to subscribe to. + std::string participant_identity; + + /// Pixel format requested from the SDK. I420 costs no conversion on + /// either side. + PixelFormat video_format = PixelFormat::I420; + + /// Ring-buffer depth for decoded video. Non-zero means the SDK drops the + /// OLDEST frame when the queue is full, which is the structural answer to + /// the stale-frame-after-publisher-swap bug that motivated this plugin + /// (see the design doc's Approach section): a stalled consumer can only + /// ever fall this far behind, and what it then sees is the newest frame, + /// not a backlog. + std::size_t video_queue_capacity = 3; + + /// Same for audio. A little deeper because audio frames are 10ms each. + std::size_t audio_queue_capacity = 20; + + bool subscribe_audio = true; + + /// How long connect() waits for the room to come up before giving up. + int connect_timeout_ms = 15000; +}; + +/// Wraps livekit::Room for exactly one subscribed slot. +/// +/// Threading contract, which the OBS adapter depends on: +/// - connect() and disconnect() are blocking and must be called from an +/// ordinary thread. They must NOT be called from inside a handler this +/// class invokes: the SDK documents that Room::disconnect() deadlocks if +/// called from a room event callback, and Room's own callback registration +/// is not re-entrant either. +/// - The video and audio handlers are invoked on dedicated reader threads, +/// one per track. Frame pointers are valid only for the duration of the +/// call. +/// - The state handler is invoked from whichever thread observed the +/// change. It must not block and must not call back into this object. +/// - All handlers must be installed before connect(); they are not +/// synchronised against a running session. +class LiveKitSession { +public: + LiveKitSession(); + ~LiveKitSession(); + + LiveKitSession(const LiveKitSession &) = delete; + LiveKitSession &operator=(const LiveKitSession &) = delete; + + void setVideoHandler(VideoFrameHandler handler); + void setAudioHandler(AudioFrameHandler handler); + void setStateHandler(SessionStateHandler handler); + + /// Connect and start subscribing. Returns true once the room is up; the + /// selected slot's tracks may still arrive later (or not at all, if the + /// camera is dark), which is reported through hasVideo()/the state + /// handler rather than as a connect failure. + bool connect(const SessionConfig &config); + + /// Tear everything down. Safe to call when never connected, and safe to + /// call twice. + void disconnect(); + + SessionState state() const; + std::string stateDetail() const; + bool hasVideo() const; + bool hasAudio() const; + /// True when connected but the slot is not publishing: the source should + /// show its placeholder, not an error. + bool waitingForCamera() const; + + /// Monotonic counters, for logging and for the adapter to tell "connected + /// but silent" from "never started". + std::uint64_t videoFrameCount() const; + std::uint64_t audioFrameCount() const; + + /// Process-wide SDK init/teardown. Reference-counted, so several sources + /// can each hold one. The OBS adapter calls these from obs_module_load / + /// obs_module_unload. + static void globalInitialize(); + static void globalShutdown(); + +private: + struct Impl; + std::unique_ptr impl_; +}; + +} // namespace stplugin diff --git a/core/include/stplugin/session_types.h b/core/include/stplugin/session_types.h new file mode 100644 index 0000000..758e6d0 --- /dev/null +++ b/core/include/stplugin/session_types.h @@ -0,0 +1,187 @@ +/* +streamer-tools OBS Camera Plugin - session types and pure session logic +Copyright (C) 2026 CyberCoveLLC + +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 +*/ + +#pragma once + +// Everything in this header is deliberately free of both LiveKit and OBS +// types. That is what makes the session's decision-making unit-testable +// headlessly: the parts of LiveKitSession that can go wrong without a server +// -- which state a sequence of room events leaves us in, whether a given +// published track is the one we want, and whether a frame's geometry is +// self-consistent -- all live here, and LiveKitSession is the (much thinner) +// piece that wires real SDK callbacks into them. + +#include +#include +#include +#include + +namespace stplugin { + +// --------------------------------------------------------------------------- +// Media description (mirrors of the LiveKit enums, converted at the boundary) +// --------------------------------------------------------------------------- + +enum class MediaKind { Unknown, Audio, Video }; + +enum class MediaSource { Unknown, Camera, Microphone, Screenshare, ScreenshareAudio, Other }; + +/// Pixel formats this plugin is willing to receive. I420 is the default +/// because it is what a WebRTC decoder produces and what OBS accepts +/// natively, so neither side pays for a conversion. +enum class PixelFormat { I420, NV12, BGRA }; + +const char *describePixelFormat(PixelFormat format); + +/// Number of planes a format uses (3 for I420, 2 for NV12, 1 for BGRA). +int planeCount(PixelFormat format); + +/// Total bytes a tightly-packed frame of this format and geometry occupies. +/// Returns 0 for non-positive dimensions. Used to reject a frame whose +/// buffer does not match its claimed size before its pointers reach OBS. +std::size_t expectedFrameBytes(PixelFormat format, int width, int height); + +// --------------------------------------------------------------------------- +// Track selection +// --------------------------------------------------------------------------- + +/// Does this published track belong to the slot we were asked to show, and is +/// it the camera (never the screenshare)? +/// +/// streamer-tools publishes cameras as Track.Source.Camera and screenshares +/// as Track.Source.ScreenShare (apps/web/src/avatar/publish.ts), so the +/// source is the reliable discriminator. A track that reports no source at +/// all is accepted on kind alone rather than dropped, since an unknown source +/// on a video track from the right participant is far more likely to be a +/// camera than anything else. +bool isWantedVideoTrack(const std::string &wanted_identity, const std::string &track_identity, MediaKind kind, + MediaSource source); + +/// Same, for the slot's microphone. +bool isWantedAudioTrack(const std::string &wanted_identity, const std::string &track_identity, MediaKind kind, + MediaSource source); + +// --------------------------------------------------------------------------- +// Session state +// --------------------------------------------------------------------------- + +enum class SessionState { + /// Never asked to connect, or fully torn down. + Idle, + /// connect() is in flight. + Connecting, + /// Signalling is up. Says nothing about whether video is arriving -- + /// that is what hasVideo() is for. + Connected, + /// The SDK is re-establishing the connection on its own. + Reconnecting, + /// The room ended: either we disconnected, or the server did. + Disconnected, + /// connect() failed, or the session died in a way retrying will not fix + /// (a rejected token, a duplicate identity). + Failed, +}; + +const char *describeSessionState(SessionState state); + +/// The pure state machine behind LiveKitSession. Not thread-safe on its own; +/// LiveKitSession owns the lock. +/// +/// It exists separately so the transitions that matter operationally -- a +/// publisher swap must not read as a failure, a reconnect must not read as a +/// fresh connection, a failure detail must survive until the next connect -- +/// can be tested without a LiveKit server. +class SessionStateMachine { +public: + SessionState state() const { return state_; } + + /// Human-readable reason for the current state. Empty when there is + /// nothing to say. Never contains a token or read key. + const std::string &detail() const { return detail_; } + + /// Whether a video track is currently attached and delivering. + bool hasVideo() const { return has_video_; } + bool hasAudio() const { return has_audio_; } + + /// True when the source should be showing its "waiting for camera" + /// placeholder rather than an error: we are up, the slot just isn't live. + bool waitingForCamera() const; + + void onConnectRequested(); + void onConnectSucceeded(); + void onConnectFailed(const std::string &reason); + void onReconnecting(); + void onReconnected(); + /// The server (or the SDK) ended the room. `fatal` distinguishes a reason + /// that retrying cannot fix from an ordinary drop. + void onRoomEnded(const std::string &reason, bool fatal); + void onLocalDisconnect(); + + void onVideoAttached(); + void onVideoDetached(); + void onAudioAttached(); + void onAudioDetached(); + +private: + SessionState state_ = SessionState::Idle; + std::string detail_; + bool has_video_ = false; + bool has_audio_ = false; +}; + +// --------------------------------------------------------------------------- +// Frames handed to the OBS adapter +// --------------------------------------------------------------------------- + +struct VideoPlane { + const std::uint8_t *data = nullptr; + std::uint32_t stride = 0; + std::uint32_t size = 0; +}; + +/// A decoded video frame. All pointers are owned by the SDK and are valid +/// only for the duration of the callback -- copy or consume synchronously. +struct VideoFrameData { + int width = 0; + int height = 0; + PixelFormat format = PixelFormat::I420; + const std::uint8_t *data = nullptr; + std::size_t size = 0; + VideoPlane planes[4]; + int plane_count = 0; + /// WebRTC capture-time timestamp, microseconds. + std::int64_t timestamp_us = 0; +}; + +/// Interleaved int16 PCM. client-sdk-cpp's AudioFrameCallback carries no +/// timestamp, so the adapter stamps arrival time itself -- see the design +/// doc's "Audio/video sync verification" note, which flags that as an +/// assumption to check on real hardware rather than a guarantee. +struct AudioFrameData { + const std::int16_t *samples = nullptr; + std::size_t sample_count = 0; + int sample_rate = 0; + int channels = 0; + int samples_per_channel = 0; +}; + +using VideoFrameHandler = std::function; +using AudioFrameHandler = std::function; +using SessionStateHandler = std::function; + +} // namespace stplugin diff --git a/core/src/session.cpp b/core/src/session.cpp new file mode 100644 index 0000000..2682cf1 --- /dev/null +++ b/core/src/session.cpp @@ -0,0 +1,733 @@ +/* +streamer-tools OBS Camera Plugin - LiveKit session wrapper +Copyright (C) 2026 CyberCoveLLC + +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 +*/ + +#include "stplugin/session.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace stplugin { + +namespace { + +// --- LiveKit <-> plugin type conversion ------------------------------------ + +MediaKind toMediaKind(livekit::TrackKind kind) +{ + switch (kind) { + case livekit::TrackKind::KIND_AUDIO: return MediaKind::Audio; + case livekit::TrackKind::KIND_VIDEO: return MediaKind::Video; + case livekit::TrackKind::KIND_UNKNOWN: break; + } + return MediaKind::Unknown; +} + +MediaSource toMediaSource(livekit::TrackSource source) +{ + switch (source) { + case livekit::TrackSource::SOURCE_CAMERA: return MediaSource::Camera; + case livekit::TrackSource::SOURCE_MICROPHONE: return MediaSource::Microphone; + case livekit::TrackSource::SOURCE_SCREENSHARE: return MediaSource::Screenshare; + case livekit::TrackSource::SOURCE_SCREENSHARE_AUDIO: return MediaSource::ScreenshareAudio; + case livekit::TrackSource::SOURCE_UNKNOWN: break; + } + return MediaSource::Unknown; +} + +livekit::VideoBufferType toLiveKitBufferType(PixelFormat format) +{ + switch (format) { + case PixelFormat::I420: return livekit::VideoBufferType::I420; + case PixelFormat::NV12: return livekit::VideoBufferType::NV12; + case PixelFormat::BGRA: return livekit::VideoBufferType::BGRA; + } + return livekit::VideoBufferType::I420; +} + +/// Returns false when the SDK handed us a format the OBS adapter cannot +/// consume, in which case the caller converts. +bool fromLiveKitBufferType(livekit::VideoBufferType type, PixelFormat &out) +{ + switch (type) { + case livekit::VideoBufferType::I420: out = PixelFormat::I420; return true; + case livekit::VideoBufferType::NV12: out = PixelFormat::NV12; return true; + case livekit::VideoBufferType::BGRA: out = PixelFormat::BGRA; return true; + default: return false; + } +} + +/// Which disconnect reasons are worth telling the operator "this will not fix +/// itself" about. Everything else is reported as an ordinary disconnect, +/// because the SDK's own reconnect logic covers it. +bool isFatalDisconnect(livekit::DisconnectReason reason) +{ + switch (reason) { + case livekit::DisconnectReason::DuplicateIdentity: + case livekit::DisconnectReason::ParticipantRemoved: + case livekit::DisconnectReason::RoomDeleted: + case livekit::DisconnectReason::JoinFailure: + case livekit::DisconnectReason::UserRejected: + return true; + default: + return false; + } +} + +const char *describeDisconnectReason(livekit::DisconnectReason reason) +{ + switch (reason) { + case livekit::DisconnectReason::Unknown: return "connection lost"; + case livekit::DisconnectReason::ClientInitiated: return "disconnected"; + case livekit::DisconnectReason::DuplicateIdentity: return "another client joined with the same identity"; + case livekit::DisconnectReason::ServerShutdown: return "the LiveKit server is shutting down"; + case livekit::DisconnectReason::ParticipantRemoved: return "removed from the room"; + case livekit::DisconnectReason::RoomDeleted: return "the room was deleted"; + case livekit::DisconnectReason::StateMismatch: return "session could not be resumed"; + case livekit::DisconnectReason::JoinFailure: return "could not join the room (token rejected or expired?)"; + case livekit::DisconnectReason::Migration: return "migrating to another server"; + case livekit::DisconnectReason::SignalClose: return "the signalling connection closed"; + case livekit::DisconnectReason::RoomClosed: return "the room closed"; + case livekit::DisconnectReason::UserUnavailable: return "user unavailable"; + case livekit::DisconnectReason::UserRejected: return "connection rejected"; + case livekit::DisconnectReason::SipTrunkFailure: return "SIP trunk failure"; + case livekit::DisconnectReason::ConnectionTimeout: return "connection timed out"; + case livekit::DisconnectReason::MediaFailure: return "media connection failed"; + case livekit::DisconnectReason::AgentError: return "agent error"; + } + return "disconnected"; +} + +// --- Process-wide SDK lifetime --------------------------------------------- + +std::mutex &globalMutex() +{ + static std::mutex m; + return m; +} + +int &globalRefCount() +{ + static int n = 0; + return n; +} + +} // namespace + +// --------------------------------------------------------------------------- +// Impl +// --------------------------------------------------------------------------- + +struct LiveKitSession::Impl : public livekit::RoomDelegate { + enum class CommandType { AttachVideo, DetachVideo, AttachAudio, DetachAudio, Stop }; + + struct Command { + CommandType type; + std::shared_ptr track; + }; + + livekit::Room room; + SessionConfig config; + + mutable std::mutex state_mutex; + SessionStateMachine machine; + + VideoFrameHandler on_video; + AudioFrameHandler on_audio; + SessionStateHandler on_state; + + std::atomic video_frames{0}; + std::atomic audio_frames{0}; + std::atomic dropped_frames{0}; + + // Command queue. Every interaction with livekit::VideoStream / + // livekit::AudioStream happens on `worker`, never on a room event thread: + // the SDK's room callbacks run on its own event thread and blocking or + // re-entering there stalls every other event (and Room::disconnect() from + // inside one is documented to deadlock outright). + std::mutex queue_mutex; + std::condition_variable queue_cv; + std::deque queue; + std::thread worker; + bool worker_running = false; + + // Owned exclusively by the worker thread. + std::shared_ptr video_stream; + std::thread video_thread; + std::shared_ptr audio_stream; + std::thread audio_thread; + + bool connected = false; + + ~Impl() override = default; + + // --- state helpers ----------------------------------------------------- + + template void mutateState(Fn &&fn) + { + SessionState state; + std::string detail; + SessionStateHandler handler; + { + std::lock_guard guard(state_mutex); + fn(machine); + state = machine.state(); + detail = machine.detail(); + handler = on_state; + } + // Notified outside the lock: the handler is OBS adapter code and must + // never be able to deadlock against a concurrent state query. + if (handler) + handler(state, detail); + } + + void post(CommandType type, std::shared_ptr track = nullptr) + { + { + std::lock_guard guard(queue_mutex); + if (!worker_running) + return; + queue.push_back(Command{type, std::move(track)}); + } + queue_cv.notify_one(); + } + + // --- RoomDelegate ------------------------------------------------------ + + void onTrackSubscribed(livekit::Room &, const livekit::TrackSubscribedEvent &event) override + { + if (!event.participant || !event.track) + return; + const std::string identity = event.participant->identity(); + const MediaKind kind = toMediaKind(event.track->kind()); + const MediaSource source = + event.publication ? toMediaSource(event.publication->source()) : MediaSource::Unknown; + + if (isWantedVideoTrack(config.participant_identity, identity, kind, source)) + post(CommandType::AttachVideo, event.track); + else if (config.subscribe_audio && isWantedAudioTrack(config.participant_identity, identity, kind, source)) + post(CommandType::AttachAudio, event.track); + } + + void onTrackUnsubscribed(livekit::Room &, const livekit::TrackUnsubscribedEvent &event) override + { + if (!event.participant || !event.track) + return; + if (event.participant->identity() != config.participant_identity) + return; + const MediaKind kind = toMediaKind(event.track->kind()); + if (kind == MediaKind::Video) + post(CommandType::DetachVideo); + else if (kind == MediaKind::Audio) + post(CommandType::DetachAudio); + } + + void onParticipantDisconnected(livekit::Room &, const livekit::ParticipantDisconnectedEvent &event) override + { + if (!event.participant || event.participant->identity() != config.participant_identity) + return; + // The slot went away entirely. This is the placeholder state, not an + // error: the operator's room is fine, the camera just left. + post(CommandType::DetachVideo); + post(CommandType::DetachAudio); + } + + void onReconnecting(livekit::Room &, const livekit::ReconnectingEvent &) override + { + mutateState([](SessionStateMachine &m) { m.onReconnecting(); }); + } + + void onReconnected(livekit::Room &, const livekit::ReconnectedEvent &) override + { + mutateState([](SessionStateMachine &m) { m.onReconnected(); }); + } + + void onDisconnected(livekit::Room &, const livekit::DisconnectedEvent &event) override + { + const std::string reason = describeDisconnectReason(event.reason); + const bool fatal = isFatalDisconnect(event.reason); + post(CommandType::DetachVideo); + post(CommandType::DetachAudio); + mutateState([&](SessionStateMachine &m) { m.onRoomEnded(reason, fatal); }); + } + + void onRoomEos(livekit::Room &, const livekit::RoomEosEvent &) override + { + post(CommandType::DetachVideo); + post(CommandType::DetachAudio); + mutateState([](SessionStateMachine &m) { m.onRoomEnded("the room session ended", false); }); + } + + // --- worker ------------------------------------------------------------ + + void startWorker() + { + { + std::lock_guard guard(queue_mutex); + queue.clear(); + worker_running = true; + } + worker = std::thread([this] { workerLoop(); }); + } + + void stopWorker() + { + { + std::lock_guard guard(queue_mutex); + if (!worker_running) + return; + queue.push_back(Command{CommandType::Stop, nullptr}); + worker_running = false; + } + queue_cv.notify_one(); + if (worker.joinable()) + worker.join(); + } + + void workerLoop() + { + for (;;) { + Command command{CommandType::Stop, nullptr}; + { + std::unique_lock lock(queue_mutex); + queue_cv.wait(lock, [this] { return !queue.empty(); }); + command = std::move(queue.front()); + queue.pop_front(); + } + + switch (command.type) { + case CommandType::AttachVideo: + attachVideo(command.track); + break; + case CommandType::DetachVideo: + detachVideo(); + break; + case CommandType::AttachAudio: + attachAudio(command.track); + break; + case CommandType::DetachAudio: + detachAudio(); + break; + case CommandType::Stop: + detachVideo(); + detachAudio(); + return; + } + } + } + + void attachVideo(const std::shared_ptr &track) + { + if (!track) + return; + // Replacing an existing stream is the publisher-swap path: tear the + // old reader all the way down first so no frame from the previous + // publisher can arrive after the new one starts. + detachVideo(); + + livekit::VideoStream::Options options; + options.capacity = config.video_queue_capacity; + options.format = toLiveKitBufferType(config.video_format); + + std::shared_ptr stream; + try { + stream = livekit::VideoStream::fromTrack(track, options); + } catch (const std::exception &e) { + mutateState([&](SessionStateMachine &m) { + m.onRoomEnded(std::string("could not open the video stream: ") + e.what(), true); + }); + return; + } + if (!stream) + return; + + video_stream = stream; + video_thread = std::thread([this, stream] { videoReaderLoop(stream); }); + mutateState([](SessionStateMachine &m) { m.onVideoAttached(); }); + } + + void detachVideo() + { + if (video_stream) + video_stream->close(); // wakes the blocking read() + if (video_thread.joinable()) + video_thread.join(); + const bool had = static_cast(video_stream); + video_stream.reset(); + if (had) + mutateState([](SessionStateMachine &m) { m.onVideoDetached(); }); + } + + void attachAudio(const std::shared_ptr &track) + { + if (!track) + return; + detachAudio(); + + livekit::AudioStream::Options options; + options.capacity = config.audio_queue_capacity; + + std::shared_ptr stream; + try { + stream = livekit::AudioStream::fromTrack(track, options); + } catch (const std::exception &) { + // Audio is not worth failing the whole source over: a camera with + // no usable audio track is still a usable camera. + return; + } + if (!stream) + return; + + audio_stream = stream; + audio_thread = std::thread([this, stream] { audioReaderLoop(stream); }); + mutateState([](SessionStateMachine &m) { m.onAudioAttached(); }); + } + + void detachAudio() + { + if (audio_stream) + audio_stream->close(); + if (audio_thread.joinable()) + audio_thread.join(); + const bool had = static_cast(audio_stream); + audio_stream.reset(); + if (had) + mutateState([](SessionStateMachine &m) { m.onAudioDetached(); }); + } + + void videoReaderLoop(std::shared_ptr stream) + { + VideoFrameHandler handler; + { + std::lock_guard guard(state_mutex); + handler = on_video; + } + + livekit::VideoFrameEvent event; + while (stream->read(event)) { + if (!handler) + continue; + deliverVideoFrame(event, handler); + } + } + + void deliverVideoFrame(livekit::VideoFrameEvent &event, const VideoFrameHandler &handler) + { + PixelFormat format; + const livekit::VideoFrame *frame = &event.frame; + livekit::VideoFrame converted; + + if (!fromLiveKitBufferType(frame->type(), format)) { + // The SDK gave us something the adapter cannot hand to OBS. + // convert() is a full CPU repack, so this is a fallback, not the + // normal path -- the normal path is the format we asked for. + try { + converted = frame->convert(toLiveKitBufferType(config.video_format)); + } catch (const std::exception &) { + dropped_frames.fetch_add(1); + return; + } + frame = &converted; + format = config.video_format; + } + + const int width = frame->width(); + const int height = frame->height(); + const std::size_t expected = expectedFrameBytes(format, width, height); + if (expected == 0 || frame->dataSize() < expected) { + // Geometry that does not match the buffer would make OBS read off + // the end of it. Drop rather than trust. + dropped_frames.fetch_add(1); + return; + } + + VideoFrameData out; + out.width = width; + out.height = height; + out.format = format; + out.data = frame->data(); + out.size = frame->dataSize(); + out.timestamp_us = event.timestamp_us; + + const std::vector planes = frame->planeInfos(); + const int wanted_planes = planeCount(format); + int count = 0; + for (const livekit::VideoPlaneInfo &plane : planes) { + if (count >= 4) + break; + out.planes[count].data = reinterpret_cast(plane.data_ptr); + out.planes[count].stride = plane.stride; + out.planes[count].size = plane.size; + ++count; + } + if (count == 0 && wanted_planes == 1) { + // planeInfos() documents that packed formats may return an empty + // list rather than one plane. Synthesise it from the frame buffer + // instead of dropping a perfectly good BGRA frame. + out.planes[0].data = frame->data(); + out.planes[0].stride = static_cast(width) * 4u; + out.planes[0].size = static_cast(frame->dataSize()); + count = 1; + } + if (count != wanted_planes) { + dropped_frames.fetch_add(1); + return; + } + out.plane_count = count; + + video_frames.fetch_add(1); + handler(out); + } + + void audioReaderLoop(std::shared_ptr stream) + { + AudioFrameHandler handler; + { + std::lock_guard guard(state_mutex); + handler = on_audio; + } + + livekit::AudioFrameEvent event; + while (stream->read(event)) { + if (!handler) + continue; + const livekit::AudioFrame &frame = event.frame; + if (frame.numChannels() <= 0 || frame.samplesPerChannel() <= 0 || frame.sampleRate() <= 0) + continue; + AudioFrameData out; + out.samples = frame.data().data(); + out.sample_count = frame.totalSamples(); + out.sample_rate = frame.sampleRate(); + out.channels = frame.numChannels(); + out.samples_per_channel = frame.samplesPerChannel(); + audio_frames.fetch_add(1); + handler(out); + } + } + + /// After connect(), the target slot may already be in the room with its + /// tracks subscribed, in which case no onTrackSubscribed event is coming. + /// Sweep what is already there so a source added mid-show shows video + /// immediately instead of waiting for the publisher to republish. + void attachExistingTracks() + { + auto participant = room.remoteParticipant(config.participant_identity).lock(); + if (!participant) + return; + const std::string identity = participant->identity(); + for (const auto &entry : participant->trackPublications()) { + const std::shared_ptr &publication = entry.second; + if (!publication) + continue; + const std::shared_ptr track = publication->track(); + if (!track) + continue; // published but not subscribed yet + const MediaKind kind = toMediaKind(track->kind()); + const MediaSource source = toMediaSource(publication->source()); + if (isWantedVideoTrack(config.participant_identity, identity, kind, source)) + post(CommandType::AttachVideo, track); + else if (config.subscribe_audio && isWantedAudioTrack(config.participant_identity, identity, kind, source)) + post(CommandType::AttachAudio, track); + } + } +}; + +// --------------------------------------------------------------------------- +// LiveKitSession +// --------------------------------------------------------------------------- + +LiveKitSession::LiveKitSession() : impl_(new Impl()) {} + +LiveKitSession::~LiveKitSession() +{ + disconnect(); +} + +void LiveKitSession::setVideoHandler(VideoFrameHandler handler) +{ + std::lock_guard guard(impl_->state_mutex); + impl_->on_video = std::move(handler); +} + +void LiveKitSession::setAudioHandler(AudioFrameHandler handler) +{ + std::lock_guard guard(impl_->state_mutex); + impl_->on_audio = std::move(handler); +} + +void LiveKitSession::setStateHandler(SessionStateHandler handler) +{ + std::lock_guard guard(impl_->state_mutex); + impl_->on_state = std::move(handler); +} + +bool LiveKitSession::connect(const SessionConfig &config) +{ + if (impl_->connected) + disconnect(); + + impl_->config = config; + impl_->video_frames.store(0); + impl_->audio_frames.store(0); + impl_->dropped_frames.store(0); + + impl_->mutateState([](SessionStateMachine &m) { m.onConnectRequested(); }); + + if (config.ws_url.empty() || config.token.empty() || config.participant_identity.empty()) { + impl_->mutateState( + [](SessionStateMachine &m) { m.onConnectFailed("missing LiveKit URL, token or camera selection"); }); + return false; + } + + impl_->startWorker(); + + livekit::RoomOptions options; + // auto_subscribe is what makes track_subscribed events (and therefore any + // media at all) happen; the SDK is emphatic about this. + options.auto_subscribe = true; + options.dynacast = false; + // This client never publishes, so a single peer connection is all it + // needs. + options.single_peer_connection = true; + options.connect_timeout = std::chrono::milliseconds(config.connect_timeout_ms); + + impl_->room.setDelegate(impl_.get()); + + bool ok = false; + try { + ok = impl_->room.connect(config.ws_url, config.token, options); + } catch (const std::exception &e) { + ok = false; + impl_->mutateState([&](SessionStateMachine &m) { m.onConnectFailed(e.what()); }); + impl_->stopWorker(); + impl_->room.setDelegate(nullptr); + return false; + } + + if (!ok) { + impl_->mutateState([](SessionStateMachine &m) { + m.onConnectFailed("could not connect to LiveKit (check the server URL, or the token may have expired)"); + }); + impl_->stopWorker(); + impl_->room.setDelegate(nullptr); + return false; + } + + impl_->connected = true; + impl_->mutateState([](SessionStateMachine &m) { m.onConnectSucceeded(); }); + impl_->attachExistingTracks(); + return true; +} + +void LiveKitSession::disconnect() +{ + if (!impl_) + return; + + // Order matters: stop the readers first so nothing is mid-read on a + // stream the room is about to tear down, then disconnect the room, then + // drop the delegate so no event can arrive at a half-destroyed object. + impl_->stopWorker(); + + if (impl_->connected) { + impl_->connected = false; + try { + impl_->room.disconnect(livekit::DisconnectReason::ClientInitiated); + } catch (const std::exception &) { + // Best effort: a failed graceful disconnect must not stop the + // OBS source from being destroyed. + } + impl_->mutateState([](SessionStateMachine &m) { m.onLocalDisconnect(); }); + } + + impl_->room.setDelegate(nullptr); +} + +SessionState LiveKitSession::state() const +{ + std::lock_guard guard(impl_->state_mutex); + return impl_->machine.state(); +} + +std::string LiveKitSession::stateDetail() const +{ + std::lock_guard guard(impl_->state_mutex); + return impl_->machine.detail(); +} + +bool LiveKitSession::hasVideo() const +{ + std::lock_guard guard(impl_->state_mutex); + return impl_->machine.hasVideo(); +} + +bool LiveKitSession::hasAudio() const +{ + std::lock_guard guard(impl_->state_mutex); + return impl_->machine.hasAudio(); +} + +bool LiveKitSession::waitingForCamera() const +{ + std::lock_guard guard(impl_->state_mutex); + return impl_->machine.waitingForCamera(); +} + +std::uint64_t LiveKitSession::videoFrameCount() const +{ + return impl_->video_frames.load(); +} + +std::uint64_t LiveKitSession::audioFrameCount() const +{ + return impl_->audio_frames.load(); +} + +void LiveKitSession::globalInitialize() +{ + std::lock_guard guard(globalMutex()); + if (globalRefCount()++ == 0) + livekit::initialize(livekit::LogLevel::Warn); +} + +void LiveKitSession::globalShutdown() +{ + std::lock_guard guard(globalMutex()); + if (globalRefCount() > 0 && --globalRefCount() == 0) + livekit::shutdown(); +} + +} // namespace stplugin diff --git a/core/src/session_types.cpp b/core/src/session_types.cpp new file mode 100644 index 0000000..03f7c09 --- /dev/null +++ b/core/src/session_types.cpp @@ -0,0 +1,185 @@ +/* +streamer-tools OBS Camera Plugin - session types and pure session logic +Copyright (C) 2026 CyberCoveLLC + +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 +*/ + +#include "stplugin/session_types.h" + +namespace stplugin { + +const char *describePixelFormat(PixelFormat format) +{ + switch (format) { + case PixelFormat::I420: return "I420"; + case PixelFormat::NV12: return "NV12"; + case PixelFormat::BGRA: return "BGRA"; + } + return "unknown"; +} + +int planeCount(PixelFormat format) +{ + switch (format) { + case PixelFormat::I420: return 3; + case PixelFormat::NV12: return 2; + case PixelFormat::BGRA: return 1; + } + return 0; +} + +std::size_t expectedFrameBytes(PixelFormat format, int width, int height) +{ + if (width <= 0 || height <= 0) + return 0; + const std::size_t w = static_cast(width); + const std::size_t h = static_cast(height); + // Chroma planes round up, which is what libyuv/WebRTC do for odd sizes. + const std::size_t cw = (w + 1) / 2; + const std::size_t ch = (h + 1) / 2; + switch (format) { + case PixelFormat::I420: return w * h + 2 * cw * ch; + case PixelFormat::NV12: return w * h + 2 * cw * ch; + case PixelFormat::BGRA: return w * h * 4; + } + return 0; +} + +bool isWantedVideoTrack(const std::string &wanted_identity, const std::string &track_identity, MediaKind kind, + MediaSource source) +{ + if (wanted_identity.empty() || track_identity != wanted_identity) + return false; + if (kind != MediaKind::Video) + return false; + return source == MediaSource::Camera || source == MediaSource::Unknown; +} + +bool isWantedAudioTrack(const std::string &wanted_identity, const std::string &track_identity, MediaKind kind, + MediaSource source) +{ + if (wanted_identity.empty() || track_identity != wanted_identity) + return false; + if (kind != MediaKind::Audio) + return false; + return source == MediaSource::Microphone || source == MediaSource::Unknown; +} + +const char *describeSessionState(SessionState state) +{ + switch (state) { + case SessionState::Idle: return "idle"; + case SessionState::Connecting: return "connecting"; + case SessionState::Connected: return "connected"; + case SessionState::Reconnecting: return "reconnecting"; + case SessionState::Disconnected: return "disconnected"; + case SessionState::Failed: return "failed"; + } + return "unknown"; +} + +bool SessionStateMachine::waitingForCamera() const +{ + return (state_ == SessionState::Connected || state_ == SessionState::Reconnecting) && !has_video_; +} + +void SessionStateMachine::onConnectRequested() +{ + state_ = SessionState::Connecting; + // A new attempt clears the previous failure reason, so a stale message + // can never be shown alongside a fresh, healthy connection. + detail_.clear(); + has_video_ = false; + has_audio_ = false; +} + +void SessionStateMachine::onConnectSucceeded() +{ + state_ = SessionState::Connected; + detail_.clear(); +} + +void SessionStateMachine::onConnectFailed(const std::string &reason) +{ + state_ = SessionState::Failed; + detail_ = reason; + has_video_ = false; + has_audio_ = false; +} + +void SessionStateMachine::onReconnecting() +{ + // Only meaningful from a live session; a reconnect notification after we + // already gave up must not resurrect the session. + if (state_ != SessionState::Connected && state_ != SessionState::Reconnecting) + return; + state_ = SessionState::Reconnecting; + detail_ = "reconnecting"; + // Tracks are re-subscribed on the other side of a reconnect; the SDK will + // tell us when they are back. + has_video_ = false; + has_audio_ = false; +} + +void SessionStateMachine::onReconnected() +{ + if (state_ != SessionState::Reconnecting) + return; + state_ = SessionState::Connected; + detail_.clear(); +} + +void SessionStateMachine::onRoomEnded(const std::string &reason, bool fatal) +{ + if (state_ == SessionState::Idle || state_ == SessionState::Disconnected) + return; + state_ = fatal ? SessionState::Failed : SessionState::Disconnected; + detail_ = reason; + has_video_ = false; + has_audio_ = false; +} + +void SessionStateMachine::onLocalDisconnect() +{ + state_ = SessionState::Disconnected; + detail_.clear(); + has_video_ = false; + has_audio_ = false; +} + +void SessionStateMachine::onVideoAttached() +{ + has_video_ = true; +} + +void SessionStateMachine::onVideoDetached() +{ + // A publisher swap (the motivating bug) shows up here: the old track goes + // away and a new one arrives moments later. That is a placeholder state, + // never an error state -- the connection itself is untouched. + has_video_ = false; +} + +void SessionStateMachine::onAudioAttached() +{ + has_audio_ = true; +} + +void SessionStateMachine::onAudioDetached() +{ + has_audio_ = false; +} + +} // namespace stplugin diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index 343c97c..313ef78 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -17,6 +17,14 @@ endfunction() stplugin_add_test(test_core) stplugin_add_test(test_json) stplugin_add_test(test_api_client) +stplugin_add_test(test_session) + +# End-to-end against a REAL LiveKit room: publishes a synthetic camera with +# the same SDK and subscribes to it through LiveKitSession. Skips (exit 0) +# unless STPLUGIN_IT_* is set, so the three build runners -- which have no +# LiveKit server -- stay green. See scripts/livekit-dev-room.py. +stplugin_add_test(test_integration_livekit) +set_tests_properties(test_integration_livekit PROPERTIES TIMEOUT 300) # Smoke test for the LiveKit SDK link: initialize()/shutdown() must succeed # in-process. This is the cheapest possible proof that LiveKit::livekit is diff --git a/core/tests/test_integration_livekit.cpp b/core/tests/test_integration_livekit.cpp new file mode 100644 index 0000000..3b3c38d --- /dev/null +++ b/core/tests/test_integration_livekit.cpp @@ -0,0 +1,370 @@ +/* +streamer-tools OBS Camera Plugin - LiveKit end-to-end integration test +Copyright (C) 2026 CyberCoveLLC + +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 +*/ + +// 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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(kWidth) * kHeight; + std::memset(data, 16, luma); + const int band = (tick * 7) % kHeight; + std::memset(data + static_cast(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 &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 video_source; + std::shared_ptr video_track; + std::shared_ptr audio_source; + std::shared_ptr audio_track; + std::thread pump; + std::atomic 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(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(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 pcm(480, 0); // 10ms of 48kHz mono + while (!stop.load()) { + if (video_source) { + livekit::VideoFrame frame = makeFrame(tick); + video_source->captureFrame(frame, static_cast(tick) * 33333); + } + if (audio_source) { + for (std::size_t i = 0; i < pcm.size(); ++i) + pcm[i] = static_cast(((tick * 480 + static_cast(i)) % 100) * 100); + livekit::AudioFrame audio(pcm, 48000, 1, static_cast(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_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 video_frames{0}; + std::atomic audio_frames{0}; + std::atomic bad_frames{0}; + std::atomic full_size_frames{0}; + std::atomic last_width{0}; + std::atomic last_height{0}; + std::atomic last_planes{0}; + std::atomic last_timestamp{0}; + std::atomic last_sample_rate{0}; + std::atomic 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((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(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(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 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"); +} diff --git a/core/tests/test_session.cpp b/core/tests/test_session.cpp new file mode 100644 index 0000000..dbe469f --- /dev/null +++ b/core/tests/test_session.cpp @@ -0,0 +1,355 @@ +/* +streamer-tools OBS Camera Plugin - session wrapper tests +Copyright (C) 2026 CyberCoveLLC + +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 +*/ + +// What is and is not covered here, stated plainly because it matters: +// +// COVERED headlessly -- the session's own decision-making: the state +// machine's transitions (including the publisher-swap and reconnect paths +// that motivated this plugin), track selection, frame geometry validation, +// and the real connect() failure paths against the real SDK (bad URL, +// unreachable host, garbage token). +// +// NOT COVERED here -- anything that needs a LiveKit server to answer: +// a successful connect, actual subscription, and actual decoded frames +// reaching the handlers. Those can only be verified against a real room, +// and the design doc's Testing section puts that in the integration-test / +// manual-sign-off bucket. + +#include +#include +#include +#include + +#include "stplugin/session.h" +#include "stplugin/session_types.h" +#include "test_util.h" + +using namespace stplugin; + +namespace { + +// --------------------------------------------------------------------------- +// Pure logic +// --------------------------------------------------------------------------- + +void testFrameGeometry() +{ + ST_ASSERT_EQ(planeCount(PixelFormat::I420), 3); + ST_ASSERT_EQ(planeCount(PixelFormat::NV12), 2); + ST_ASSERT_EQ(planeCount(PixelFormat::BGRA), 1); + + // 1280x720 I420: 921600 luma + 2 * 230400 chroma. + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 1280, 720), std::size_t(1382400)); + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::NV12, 1280, 720), std::size_t(1382400)); + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::BGRA, 1280, 720), std::size_t(3686400)); + + // Odd dimensions round the chroma planes up, the way libyuv does. + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 3, 3), std::size_t(9 + 2 * 4)); + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 1, 1), std::size_t(1 + 2)); + + // Degenerate geometry is 0, which the reader treats as "drop the frame". + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 0, 720), std::size_t(0)); + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 1280, 0), std::size_t(0)); + ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, -1, -1), std::size_t(0)); + + ST_ASSERT_EQ(std::string(describePixelFormat(PixelFormat::I420)), std::string("I420")); +} + +void testTrackSelection() +{ + const std::string want = "cam1"; + + // The camera we asked for. + ST_ASSERT(isWantedVideoTrack(want, "cam1", MediaKind::Video, MediaSource::Camera)); + // A video track with no declared source is taken on kind alone. + ST_ASSERT(isWantedVideoTrack(want, "cam1", MediaKind::Video, MediaSource::Unknown)); + + // Someone else's camera. + ST_ASSERT(!isWantedVideoTrack(want, "cam2", MediaKind::Video, MediaSource::Camera)); + // The right participant's screenshare is explicitly NOT the camera -- + // streamer-tools publishes those as separate sources. + ST_ASSERT(!isWantedVideoTrack(want, "cam1", MediaKind::Video, MediaSource::Screenshare)); + // Their microphone is not a video track. + ST_ASSERT(!isWantedVideoTrack(want, "cam1", MediaKind::Audio, MediaSource::Microphone)); + // No selection means nothing matches -- never "the first thing we see". + ST_ASSERT(!isWantedVideoTrack("", "cam1", MediaKind::Video, MediaSource::Camera)); + ST_ASSERT(!isWantedVideoTrack("", "", MediaKind::Video, MediaSource::Camera)); + + ST_ASSERT(isWantedAudioTrack(want, "cam1", MediaKind::Audio, MediaSource::Microphone)); + ST_ASSERT(isWantedAudioTrack(want, "cam1", MediaKind::Audio, MediaSource::Unknown)); + ST_ASSERT(!isWantedAudioTrack(want, "cam1", MediaKind::Audio, MediaSource::ScreenshareAudio)); + ST_ASSERT(!isWantedAudioTrack(want, "cam1", MediaKind::Video, MediaSource::Camera)); + ST_ASSERT(!isWantedAudioTrack(want, "other", MediaKind::Audio, MediaSource::Microphone)); +} + +void testStateMachineHappyPath() +{ + SessionStateMachine m; + ST_ASSERT(m.state() == SessionState::Idle); + ST_ASSERT(!m.hasVideo()); + ST_ASSERT(!m.waitingForCamera()); + + m.onConnectRequested(); + ST_ASSERT(m.state() == SessionState::Connecting); + // Connecting is not "waiting for camera": the placeholder belongs to a + // live connection with a dark slot, not to a connection in progress. + ST_ASSERT(!m.waitingForCamera()); + + m.onConnectSucceeded(); + ST_ASSERT(m.state() == SessionState::Connected); + ST_ASSERT(m.waitingForCamera()); + + m.onVideoAttached(); + ST_ASSERT(m.hasVideo()); + ST_ASSERT(!m.waitingForCamera()); + + m.onAudioAttached(); + ST_ASSERT(m.hasAudio()); + + m.onLocalDisconnect(); + ST_ASSERT(m.state() == SessionState::Disconnected); + ST_ASSERT(!m.hasVideo()); + ST_ASSERT(!m.hasAudio()); +} + +void testPublisherSwapIsNotAnError() +{ + // The motivating bug: a slot's publisher restarts mid-show. That must + // read as "waiting for camera", never as a failure, and the connection + // state must not move at all. + SessionStateMachine m; + m.onConnectRequested(); + m.onConnectSucceeded(); + m.onVideoAttached(); + + m.onVideoDetached(); + ST_ASSERT(m.state() == SessionState::Connected); + ST_ASSERT(!m.hasVideo()); + ST_ASSERT(m.waitingForCamera()); + ST_ASSERT(m.detail().empty()); + + m.onVideoAttached(); + ST_ASSERT(m.state() == SessionState::Connected); + ST_ASSERT(m.hasVideo()); + ST_ASSERT(!m.waitingForCamera()); +} + +void testReconnect() +{ + SessionStateMachine m; + m.onConnectRequested(); + m.onConnectSucceeded(); + m.onVideoAttached(); + + m.onReconnecting(); + ST_ASSERT(m.state() == SessionState::Reconnecting); + // Tracks are re-subscribed on the far side, so video is not live yet. + ST_ASSERT(!m.hasVideo()); + ST_ASSERT(m.waitingForCamera()); + ST_ASSERT_EQ(m.detail(), std::string("reconnecting")); + + m.onReconnected(); + ST_ASSERT(m.state() == SessionState::Connected); + ST_ASSERT(m.detail().empty()); + + // A stray reconnect notification after a hard failure must not resurrect + // the session. + SessionStateMachine dead; + dead.onConnectRequested(); + dead.onConnectFailed("token rejected"); + dead.onReconnecting(); + ST_ASSERT(dead.state() == SessionState::Failed); + dead.onReconnected(); + ST_ASSERT(dead.state() == SessionState::Failed); +} + +void testFailureAndRecovery() +{ + SessionStateMachine m; + m.onConnectRequested(); + m.onConnectFailed("token rejected"); + ST_ASSERT(m.state() == SessionState::Failed); + ST_ASSERT_EQ(m.detail(), std::string("token rejected")); + ST_ASSERT(!m.waitingForCamera()); + + // A fresh attempt clears the stale reason, so a healthy connection can + // never be shown next to the previous failure's message. + m.onConnectRequested(); + ST_ASSERT(m.detail().empty()); + m.onConnectSucceeded(); + ST_ASSERT(m.state() == SessionState::Connected); + ST_ASSERT(m.detail().empty()); + + // A fatal room end (duplicate identity, token rejected) is Failed; an + // ordinary drop is Disconnected. + SessionStateMachine fatal; + fatal.onConnectRequested(); + fatal.onConnectSucceeded(); + fatal.onRoomEnded("another client joined with the same identity", true); + ST_ASSERT(fatal.state() == SessionState::Failed); + + SessionStateMachine dropped; + dropped.onConnectRequested(); + dropped.onConnectSucceeded(); + dropped.onRoomEnded("the signalling connection closed", false); + ST_ASSERT(dropped.state() == SessionState::Disconnected); + + // Room-ended events after we are already down are ignored, so a late + // event cannot overwrite the reason the operator needs to see. + SessionStateMachine idle; + idle.onRoomEnded("stray", true); + ST_ASSERT(idle.state() == SessionState::Idle); +} + +// --------------------------------------------------------------------------- +// Real SDK, failure paths only (no LiveKit server available headlessly) +// --------------------------------------------------------------------------- + +void testConnectRejectsIncompleteConfig() +{ + LiveKitSession session; + std::atomic state_calls{0}; + session.setStateHandler([&](SessionState, const std::string &) { state_calls.fetch_add(1); }); + + SessionConfig config; + config.ws_url = ""; + config.token = "t"; + config.participant_identity = "cam1"; + ST_ASSERT(!session.connect(config)); + ST_ASSERT(session.state() == SessionState::Failed); + ST_ASSERT(!session.stateDetail().empty()); + + config.ws_url = "ws://127.0.0.1:1"; + config.token = ""; + ST_ASSERT(!session.connect(config)); + ST_ASSERT(session.state() == SessionState::Failed); + + config.token = "t"; + config.participant_identity = ""; + ST_ASSERT(!session.connect(config)); + ST_ASSERT(session.state() == SessionState::Failed); + + // The state handler fired for each attempt (Connecting + Failed). + ST_ASSERT(state_calls.load() >= 6); + + // Frame counters stay at zero and nothing crashes on teardown. + ST_ASSERT_EQ(session.videoFrameCount(), std::uint64_t(0)); + ST_ASSERT_EQ(session.audioFrameCount(), std::uint64_t(0)); + session.disconnect(); + session.disconnect(); // idempotent + ST_ASSERT(session.state() == SessionState::Failed || session.state() == SessionState::Disconnected); +} + +void testConnectToUnreachableServerFailsCleanly() +{ + // Port 1 on loopback: nothing is listening, and the connection is + // refused immediately rather than hanging. This exercises the real + // livekit::Room::connect() failure path, with a real (garbage) token. + LiveKitSession session; + std::atomic video_frames{0}; + session.setVideoHandler([&](const VideoFrameData &) { video_frames.fetch_add(1); }); + + SessionConfig config; + config.ws_url = "ws://127.0.0.1:1"; + config.token = "not.a.real.token"; + config.participant_identity = "cam1"; + config.connect_timeout_ms = 3000; + + const auto start = std::chrono::steady_clock::now(); + const bool ok = session.connect(config); + const auto elapsed = std::chrono::steady_clock::now() - start; + + ST_ASSERT(!ok); + ST_ASSERT(session.state() == SessionState::Failed); + ST_ASSERT(!session.stateDetail().empty()); + ST_ASSERT_EQ(video_frames.load(), 0); + // Must not sit on the caller's thread indefinitely -- this runs on an OBS + // thread in the real adapter. + ST_ASSERT(std::chrono::duration_cast(elapsed).count() < 60); + + session.disconnect(); +} + +void testConnectToNonLiveKitServerFailsCleanly() +{ + // A URL that resolves and connects but is not a LiveKit signalling + // endpoint. The realistic operator mistake: pasting the app URL. + LiveKitSession session; + SessionConfig config; + config.ws_url = "ws://127.0.0.1:1/rtc"; + config.token = "eyJhbGciOiJIUzI1NiJ9.bm90YXRva2Vu.x"; + config.participant_identity = "cam1"; + config.connect_timeout_ms = 3000; + ST_ASSERT(!session.connect(config)); + ST_ASSERT(session.state() == SessionState::Failed); + session.disconnect(); +} + +void testDestroyWithoutDisconnect() +{ + // The OBS adapter destroys sources without necessarily having called + // disconnect() first (an OBS shutdown mid-connect, say). The destructor + // must join every thread it started rather than terminating. + { + LiveKitSession session; + SessionConfig config; + config.ws_url = "ws://127.0.0.1:1"; + config.token = "t"; + config.participant_identity = "cam1"; + config.connect_timeout_ms = 2000; + (void)session.connect(config); + } + ST_ASSERT(true); // reaching here at all is the assertion +} + +void testGlobalInitIsReferenceCounted() +{ + // Several OBS sources may each hold the SDK open; the last one out turns + // the lights off, and an unbalanced extra shutdown must not underflow. + LiveKitSession::globalInitialize(); + LiveKitSession::globalInitialize(); + LiveKitSession::globalShutdown(); + LiveKitSession::globalShutdown(); + LiveKitSession::globalShutdown(); // extra, must be harmless + LiveKitSession::globalInitialize(); + LiveKitSession::globalShutdown(); + ST_ASSERT(true); +} + +} // namespace + +int main() +{ + testFrameGeometry(); + testTrackSelection(); + testStateMachineHappyPath(); + testPublisherSwapIsNotAnError(); + testReconnect(); + testFailureAndRecovery(); + + LiveKitSession::globalInitialize(); + testConnectRejectsIncompleteConfig(); + testConnectToUnreachableServerFailsCleanly(); + testConnectToNonLiveKitServerFailsCleanly(); + testDestroyWithoutDisconnect(); + LiveKitSession::globalShutdown(); + + testGlobalInitIsReferenceCounted(); + + return st_test_report("session"); +} diff --git a/scripts/livekit-dev-room.py b/scripts/livekit-dev-room.py new file mode 100644 index 0000000..fa427b5 --- /dev/null +++ b/scripts/livekit-dev-room.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""Mint LiveKit JWTs for the integration test against a local dev server. + +Usage: + livekit-server --dev --bind 127.0.0.1 & + eval "$(python3 scripts/livekit-dev-room.py)" + ctest --test-dir build -R test_integration_livekit --output-on-failure + +Prints shell `export` lines for the four STPLUGIN_IT_* variables +core/tests/test_integration_livekit.cpp looks for. With no arguments it uses +`livekit-server --dev`'s built-in devkey/secret credentials. + +Standard library only (hmac + hashlib + base64) -- there is deliberately no +pip install step here, so this runs anywhere the repo is checked out. +""" + +import argparse +import base64 +import hashlib +import hmac +import json +import os +import time + + +def b64url(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def mint(api_key: str, api_secret: str, identity: str, room: str, *, publish: bool, subscribe: bool) -> str: + now = int(time.time()) + header = {"alg": "HS256", "typ": "JWT"} + claims = { + "iss": api_key, + "sub": identity, + "name": identity, + "nbf": now - 10, + "exp": now + 3600, + "video": { + "room": room, + "roomJoin": True, + "canPublish": publish, + "canSubscribe": subscribe, + "canPublishData": False, + }, + } + signing_input = f"{b64url(json.dumps(header, separators=(',', ':')).encode())}." \ + f"{b64url(json.dumps(claims, separators=(',', ':')).encode())}" + signature = hmac.new(api_secret.encode(), signing_input.encode(), hashlib.sha256).digest() + return f"{signing_input}.{b64url(signature)}" + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default=os.environ.get("LIVEKIT_URL", "ws://127.0.0.1:7880")) + parser.add_argument("--api-key", default=os.environ.get("LIVEKIT_API_KEY", "devkey")) + parser.add_argument("--api-secret", default=os.environ.get("LIVEKIT_API_SECRET", "secret")) + parser.add_argument("--room", default="obs-plugin-it") + parser.add_argument("--publisher-identity", default="cam-test") + parser.add_argument("--subscriber-identity", default="obs:obs-plugin-it:test") + args = parser.parse_args() + + publish_token = mint(args.api_key, args.api_secret, args.publisher_identity, args.room, + publish=True, subscribe=False) + # Deliberately the same grant shape the real server mints for the plugin + # (apps/server/src/obs/plugin.routes.ts -> mintCaptionsToken): subscribe + # only, never publish. + subscribe_token = mint(args.api_key, args.api_secret, args.subscriber_identity, args.room, + publish=False, subscribe=True) + + print(f'export STPLUGIN_IT_URL="{args.url}"') + print(f'export STPLUGIN_IT_PUBLISH_TOKEN="{publish_token}"') + print(f'export STPLUGIN_IT_SUBSCRIBE_TOKEN="{subscribe_token}"') + print(f'export STPLUGIN_IT_PUBLISHER_IDENTITY="{args.publisher_identity}"') + + +if __name__ == "__main__": + main()